I’m building out an OAuth2 Factory using FastAPI because my application needs to authenticate with multiple providers.
The authentication factory works well with all providers except for Mailchimp.
I don’t want to use the mailchimp_marketing client library (for a few different reasons).
When I initiate the OAuth flow in the browser, I ultimately get the following error in my callback function on token = await oauth_client.authorize_access_token(request) (exchanging the authorization code for an access token). And I’ve confirmed that "client_id=settings.MAILCHIMP_CLIENT_ID," is actually set.
"GET /oauth/mailchimp/callback?state=ABC123XYZ&code=ABC123XYZ HTTP/1.1" 500 Internal
Server Error
...
2024-07-07 16:48:30 File "/app/app/api/oauth.py", line 25, in callback
2024-07-07 16:48:30 token = await oauth_client.authorize_access_token(request)
2024-07-07 16:48:30 File "/usr/local/lib/python3.9/site-packages/authlib/integrations/starlette_client/apps.py", line 81, in authorize_access_token
2024-07-07 16:48:30 token = await self.fetch_access_token(**params, **kwargs)
2024-07-07 16:48:30 File "/usr/local/lib/python3.9/site-packages/authlib/integrations/base_client/async_app.py", line 125, in fetch_access_token
2024-07-07 16:48:30 token = await client.fetch_token(token_endpoint, **params)
2024-07-07 16:48:30 File "/usr/local/lib/python3.9/site-packages/authlib/integrations/httpx_client/oauth2_client.py", line 138, in _fetch_token
2024-07-07 16:48:30 return self.parse_response_token(resp)
2024-07-07 16:48:30 File "/usr/local/lib/python3.9/site-packages/authlib/oauth2/client.py", line 344, in parse_response_token
2024-07-07 16:48:30 raise self.oauth_error_class(
2024-07-07 16:48:30 authlib.integrations.base_client.errors.OAuthError: invalid_client: client_id parameter missing
Here’s my relevant code:
provider_factory.py
# app/services/provider_factory.py
from authlib.integrations.starlette_client import OAuth
from app.core.config import settings
from sqlalchemy.orm import Session
class ProviderFactory:
oauth = OAuth()
oauth.register(
name='mailchimp',
client_id=settings.MAILCHIMP_CLIENT_ID,
client_secret=settings.MAILCHIMP_CLIENT_SECRET,
authorize_url='https://login.mailchimp.com/oauth2/authorize',
access_token_url='https://login.mailchimp.com/oauth2/token',
api_base_url='https://login.mailchimp.com/oauth2/',
)
oauth.register(
name='another_provider',
client_id=settings.ANOTHER_CLIENT_ID,
client_secret=settings.ANOTHER_CLIENT_SECRET,
authorize_url='https://auth.another.com/oauth2/authorize',
access_token_url='https://auth.another.com/oauth2/token',
client_kwargs={'scope': '...'},
)
@staticmethod
def get_provider_service(provider_name: str, db: Session):
from app.models.provider import Provider
provider = db.query(Provider).filter_by(name=provider_name).first()
if not provider:
raise ValueError(f"No provider found with name: {provider_name}")
if provider.name == 'mailchimp':
from app.services.mailchimp import MailchimpProvider
return MailchimpProvider(db, provider)
elif provider.name == 'another':
from app.services.another import AnotherProvider
return AnotherProvider(db, provider)
else:
raise ValueError(f"Unsupported provider: {provider.name}")
@staticmethod
def get_oauth_client(provider_name: str):
return getattr(ProviderFactory.oauth, provider_name)
@staticmethod
def get_token_url(provider_name: str) -> str:
if provider_name == 'mailchimp':
return 'https://login.mailchimp.com/oauth2/token'
elif provider_name == 'another':
return 'https://auth.another.com/oauth2/token'
else:
raise ValueError("Unsupported provider")
@staticmethod
def create_provider(db: Session, provider_name: str):
return ProviderFactory.get_provider_service(provider_name, db)
oauth.py
# app/api/oauth.py
from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models.provider import Provider
from app.models.user import User
from app.models.connection import Connection
from app.core.logger import logger
from app.services.provider_factory import ProviderFactory
router = APIRouter()
@router.get("/{provider}/login")
async def login(provider: str, request: Request):
redirect_uri = request.url_for('callback', provider=provider)
return await ProviderFactory.get_oauth_client(provider).authorize_redirect(request, redirect_uri)
@router.get("/{provider}/callback")
async def callback(provider: str, request: Request, db: Session = Depends(get_db)):
logger.info(f"Provider: {provider}")
logger.info(f"Request query params: {request.query_params}")
oauth_client = ProviderFactory.get_oauth_client(provider)
token = await oauth_client.authorize_access_token(request)
if not token:
raise HTTPException(status_code=400, detail="Failed to get access token")
provider_record = db.query(Provider).filter_by(name=provider).first()
if not provider_record:
raise HTTPException(status_code=400, detail="Provider not found")
provider_service = ProviderFactory.get_provider_service(provider, db)
# Get user information from the provider
user_info = await provider_service.get_account_info(token)
logger.info(f"User info: {user_info}")
...
return {"status": "success", "provider": provider}
mailchimp.py
# app/services/mailchimp.py
import requests
from app.services.base_provider import BaseProvider
from app.models.list import List
from app.models.list_properties import ListProperties
class MailchimpProvider(BaseProvider):
tokens_expire = False
async def get_account_info(self, token: dict):
async with aiohttp.ClientSession() as session:
headers = {'Authorization': f'Bearer {token["access_token"]}'}
async with session.get('https://login.mailchimp.com/oauth2/metadata', headers=headers) as response:
return await response.json()
async def get_lists(self):
headers = {
'Authorization': f'Bearer {self.provider.access_token}'
}
response = requests.get('https://api.mailchimp.com/3.0/lists', headers=headers)
lists_data = response.json().get('lists', [])
for list_data in lists_data:
...
...
Any ideas? Hoping that I’m just missing something small.