fastapi_tools.auth.google

Authentication service for Google OAuth and session management.

Classes:

• GoogleAuthService –

  Service for Google OAuth 2.0 authentication.

• SessionStore –

  In-memory session storage.

GoogleAuthService

GoogleAuthService(
    oauth_config: GoogleOAuthConfig,
    session_config: SessionConfig,
    session_store: SessionStore,
)

Service for Google OAuth 2.0 authentication.

Initialize auth service.

Parameters:

• oauth_config (GoogleOAuthConfig) –

  Google OAuth configuration.

• session_config (SessionConfig) –

  Session management configuration.

• session_store (SessionStore) –

  Session storage backend.

Methods:

• authenticate –

  Complete authentication flow.

• create_session –

  Create a new session for authenticated user.

• exchange_code_for_tokens –

  Exchange authorization code for tokens.

• get_authorization_url –

  Generate Google OAuth authorization URL.

• get_session –

  Get session by ID.

• get_user_info –

  Get user information from Google.

• revoke_session –

  Revoke a session.

• validate_state –

  Validate OAuth state parameter.

Source code in src/fastapi_tools/auth/google.py

def __init__(
    self,
    oauth_config: GoogleOAuthConfig,
    session_config: SessionConfig,
    session_store: SessionStore,
) -> None:
    """Initialize auth service.

    Args:
        oauth_config: Google OAuth configuration.
        session_config: Session management configuration.
        session_store: Session storage backend.
    """
    self.oauth_config = oauth_config
    self.session_config = session_config
    self.session_store = session_store

authenticate async

authenticate(
    code: str, state: str, redirect_uri: str | None = None
) -> SessionData

Complete authentication flow.

Parameters:

• code (str) –

  Authorization code from Google.

• state (str) –

  State parameter for CSRF validation.

• redirect_uri (str | None, default: None ) –

  Override redirect URI. Must match the one used during login.

Returns:

• SessionData –

  SessionData for the authenticated user.

Raises:

• ValueError –

  If the state parameter is invalid.

Source code in src/fastapi_tools/auth/google.py

async def authenticate(
    self,
    code: str,
    state: str,
    redirect_uri: str | None = None,
) -> SessionData:
    """Complete authentication flow.

    Args:
        code: Authorization code from Google.
        state: State parameter for CSRF validation.
        redirect_uri: Override redirect URI. Must match the one used during login.

    Returns:
        SessionData for the authenticated user.

    Raises:
        ValueError: If the state parameter is invalid.
    """
    if not self.validate_state(state):
        msg = "Invalid state parameter"
        raise ValueError(msg)

    tokens = await self.exchange_code_for_tokens(
        code,
        redirect_uri=redirect_uri,
    )
    access_token = tokens["access_token"]
    user_info = await self.get_user_info(access_token)

    return self.create_session(user_info)

create_session

create_session(user_info: GoogleUserInfo) -> SessionData

Create a new session for authenticated user.

Source code in src/fastapi_tools/auth/google.py

def create_session(self, user_info: GoogleUserInfo) -> SessionData:
    """Create a new session for authenticated user."""
    now = datetime.now(UTC)
    session = SessionData(
        session_id=generate_session_id(),
        user_id=user_info.sub,
        email=user_info.email,
        name=user_info.name,
        picture=user_info.picture,
        created_at=now,
        expires_at=get_expiration_time(self.session_config.max_age),
    )

    self.session_store.create_session(session)
    lg.info(f"User {user_info.email} authenticated successfully")

    return session

exchange_code_for_tokens async

exchange_code_for_tokens(
    code: str, redirect_uri: str | None = None
) -> dict

Exchange authorization code for tokens.

Parameters:

• code (str) –

  Authorization code from Google.

• redirect_uri (str | None, default: None ) –

  Override redirect URI. Must match the one used during login.

Returns:

• dict –

  Token response dictionary.

Source code in src/fastapi_tools/auth/google.py

async def exchange_code_for_tokens(
    self,
    code: str,
    redirect_uri: str | None = None,
) -> dict:
    """Exchange authorization code for tokens.

    Args:
        code: Authorization code from Google.
        redirect_uri: Override redirect URI. Must match the one used during login.

    Returns:
        Token response dictionary.
    """
    data = {
        "client_id": self.oauth_config.client_id,
        "client_secret": self.oauth_config.client_secret,
        "code": code,
        "grant_type": "authorization_code",
        "redirect_uri": redirect_uri or self.oauth_config.redirect_uri,
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(GOOGLE_TOKEN_URL, data=data)
        response.raise_for_status()
        return response.json()

get_authorization_url

get_authorization_url(
    redirect_uri: str | None = None,
) -> tuple[str, str]

Generate Google OAuth authorization URL.

Parameters:

• redirect_uri (str | None, default: None ) –

  Override redirect URI (e.g. for reverse-proxy setups). If None, uses the value from oauth_config.

Returns:

• tuple[str, str] –

  Tuple of (authorization_url, state_token).

Source code in src/fastapi_tools/auth/google.py

def get_authorization_url(
    self,
    redirect_uri: str | None = None,
) -> tuple[str, str]:
    """Generate Google OAuth authorization URL.

    Args:
        redirect_uri: Override redirect URI (e.g. for reverse-proxy setups).
            If None, uses the value from oauth_config.

    Returns:
        Tuple of (authorization_url, state_token).
    """
    state = generate_state_token()
    self.session_store.store_state_token(state)

    params = {
        "client_id": self.oauth_config.client_id,
        "redirect_uri": redirect_uri or self.oauth_config.redirect_uri,
        "response_type": "code",
        "scope": " ".join(self.oauth_config.scopes),
        "state": state,
        "access_type": "offline",
        "prompt": "select_account",
    }

    auth_url = f"{GOOGLE_AUTH_URL}?{urlencode(params)}"
    return auth_url, state

get_session

get_session(session_id: str) -> SessionData | None

Get session by ID.

Source code in src/fastapi_tools/auth/google.py

def get_session(self, session_id: str) -> SessionData | None:
    """Get session by ID."""
    return self.session_store.get_session(session_id)

get_user_info async

get_user_info(access_token: str) -> GoogleUserInfo

Get user information from Google.

Source code in src/fastapi_tools/auth/google.py

async def get_user_info(self, access_token: str) -> GoogleUserInfo:
    """Get user information from Google."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            GOOGLE_USERINFO_URL,
            headers={"Authorization": f"Bearer {access_token}"},
        )
        response.raise_for_status()
        return GoogleUserInfo(**response.json())

revoke_session

revoke_session(session_id: str) -> None

Revoke a session.

Source code in src/fastapi_tools/auth/google.py

def revoke_session(self, session_id: str) -> None:
    """Revoke a session."""
    self.session_store.delete_session(session_id)

validate_state

validate_state(state: str) -> bool

Validate OAuth state parameter.

Source code in src/fastapi_tools/auth/google.py

def validate_state(self, state: str) -> bool:
    """Validate OAuth state parameter."""
    return self.session_store.validate_state_token(state)

SessionStore

SessionStore()

In-memory session storage.

For production, consider using Redis or a database.

Initialize empty session store.

Methods:

• cleanup_expired –

  Remove expired sessions and state tokens.

• create_session –

  Store a new session.

• delete_session –

  Delete a session.

• get_session –

  Retrieve a session by ID.

• store_state_token –

  Store OAuth state token for CSRF protection.

• validate_state_token –

  Validate and consume a state token.

Source code in src/fastapi_tools/auth/google.py

def __init__(self) -> None:
    """Initialize empty session store."""
    self._sessions: dict[str, SessionData] = {}
    self._state_tokens: dict[str, datetime] = {}

cleanup_expired

cleanup_expired() -> int

Remove expired sessions and state tokens.

Source code in src/fastapi_tools/auth/google.py

def cleanup_expired(self) -> int:
    """Remove expired sessions and state tokens."""
    now = datetime.now(UTC)
    removed = 0

    expired_sessions = [
        sid for sid, data in self._sessions.items() if is_expired(data.expires_at)
    ]
    for sid in expired_sessions:
        del self._sessions[sid]
        removed += 1

    expired_states = [
        state for state, exp in self._state_tokens.items() if now > exp
    ]
    for state in expired_states:
        del self._state_tokens[state]
        removed += 1

    if removed > 0:
        lg.debug(f"Cleaned up {removed} expired sessions/tokens")

    return removed

create_session

create_session(session_data: SessionData) -> None

Store a new session.

Source code in src/fastapi_tools/auth/google.py

def create_session(self, session_data: SessionData) -> None:
    """Store a new session."""
    self._sessions[session_data.session_id] = session_data
    lg.debug(f"Created session for user {session_data.email}")

delete_session

delete_session(session_id: str) -> None

Delete a session.

Source code in src/fastapi_tools/auth/google.py

def delete_session(self, session_id: str) -> None:
    """Delete a session."""
    if session_id in self._sessions:
        del self._sessions[session_id]
        lg.debug(f"Deleted session {session_id[:8]}...")

get_session

get_session(session_id: str) -> SessionData | None

Retrieve a session by ID.

Source code in src/fastapi_tools/auth/google.py

def get_session(self, session_id: str) -> SessionData | None:
    """Retrieve a session by ID."""
    session = self._sessions.get(session_id)
    if session and is_expired(session.expires_at):
        self.delete_session(session_id)
        return None
    return session

store_state_token

store_state_token(
    state: str, ttl_seconds: int = 600
) -> None

Store OAuth state token for CSRF protection.

Source code in src/fastapi_tools/auth/google.py

def store_state_token(self, state: str, ttl_seconds: int = 600) -> None:
    """Store OAuth state token for CSRF protection."""
    self._state_tokens[state] = get_expiration_time(ttl_seconds)

validate_state_token

validate_state_token(state: str) -> bool

Validate and consume a state token.

Source code in src/fastapi_tools/auth/google.py

def validate_state_token(self, state: str) -> bool:
    """Validate and consume a state token."""
    expiration = self._state_tokens.pop(state, None)
    if expiration is None:
        return False
    return not is_expired(expiration)