bruno
3bb8e87ef2
- `app/services/http_client.py` : `async with shared_client(timeout=15) as client:` remplace les 49 créations `async with httpx.AsyncClient(` de 14 fichiers (gitea ×21, providers oidc/oauth ×11, calendar ×4, automations ×3…) — le pool de connexions est réutilisé au lieu d'être recréé à chaque appel. __aexit__ no-op (le client partagé ne se ferme pas à la sortie). - Cache par (boucle d'event, kwargs) en WeakKeyDictionary : un AsyncClient n'est JAMAIS partagé entre deux loops (piège des tests « Event loop is closed ») — une boucle par test = client propre collecté avec la boucle. Clé = kwargs triés, repr() pour les valeurs non hashables (`headers=` dict → TypeError rattrapé par la suite). - Laissés délibérément : github_adapter (transport MockTransport injecté), webhook_outbound (client « own_client » fermé par la fonction). - Tests : `test_http_client_shared_and_loop_scoped` (réutilisation mêmes kwargs / cloisonné kwargs / cloisonné loop) ; le stub des webhooks patche aussi la fabrique `http_client.httpx` + purge du cache (avant : webhook_outbound.httpx patché mais la fabrique partagée créait un vrai client → réseau réel dans les tests). suite **1091/1091** · ruff OK · docs à jour
77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
"""FlowDeck — Gitea OAuth2 client."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from urllib.parse import urlencode
|
|
|
|
from app.config import settings
|
|
from app.services.http_client import shared_client
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GiteaOAuth:
|
|
"""Gitea OAuth2 client for user authentication."""
|
|
|
|
AUTHORIZE_URL = f"{settings.gitea_url}/login/oauth/authorize"
|
|
TOKEN_URL = f"{settings.gitea_url}/login/oauth/access_token"
|
|
USER_URL = f"{settings.gitea_url}/api/v1/user"
|
|
|
|
def __init__(self):
|
|
self.client_id = settings.gitea_oauth_client_id
|
|
self.client_secret = settings.gitea_oauth_client_secret
|
|
self.redirect_uri = settings.oauth_redirect_uri
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return bool(self.client_id and self.client_secret)
|
|
|
|
def get_authorize_url(self, state: str) -> str:
|
|
params = {
|
|
"client_id": self.client_id,
|
|
"redirect_uri": self.redirect_uri,
|
|
"response_type": "code",
|
|
"state": state,
|
|
}
|
|
return f"{self.AUTHORIZE_URL}?{urlencode(params)}"
|
|
|
|
async def exchange_code(self, code: str) -> dict | None:
|
|
"""Exchange authorization code for access token."""
|
|
try:
|
|
async with shared_client(timeout=15) as client:
|
|
resp = await client.post(
|
|
self.TOKEN_URL,
|
|
data={
|
|
"client_id": self.client_id,
|
|
"client_secret": self.client_secret,
|
|
"code": code,
|
|
"grant_type": "authorization_code",
|
|
"redirect_uri": self.redirect_uri,
|
|
},
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return data
|
|
except Exception as e:
|
|
logger.error("OAuth token exchange failed: %s", e)
|
|
return None
|
|
|
|
async def get_user(self, access_token: str) -> dict | None:
|
|
"""Get user info from Gitea API."""
|
|
try:
|
|
async with shared_client(timeout=10) as client:
|
|
resp = await client.get(
|
|
self.USER_URL,
|
|
headers={"Authorization": f"token {access_token}"},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception as e:
|
|
logger.error("OAuth user fetch failed: %s", e)
|
|
return None
|
|
|
|
|
|
# Singleton
|
|
gitea_oauth = GiteaOAuth()
|