Arag / app /superadmin /routers /platform_apis.py
AuthorBot
SuperAdmin QoL: persistent session, TOTP session, shared provider keys.
65a6194
Raw
History Blame Contribute Delete
2.53 kB
"""superadmin/routers/platform_apis.py — Retailer API credential management.
Routes:
GET /platform-apis
PUT /platform-apis/providers/{provider_id}
POST /platform-apis/providers/{provider_id}/disable
DELETE /platform-apis/providers/{provider_id}
"""
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_current_superadmin, get_db, get_redis
from app.schemas.superadmin import ProviderApiCredentialUpsert
from app.services.platform_api_registry import provider_scope_id
from app.services.platform_api_service import PlatformApiService
router = APIRouter()
@router.get("/platform-apis")
async def list_platform_apis(
superadmin=Depends(get_current_superadmin),
db: AsyncSession = Depends(get_db),
redis=Depends(get_redis),
):
"""List masked provider-scoped API credential status."""
return await PlatformApiService(db, redis).list_credentials()
@router.put("/platform-apis/providers/{provider_id}")
async def upsert_provider_api(
provider_id: str,
body: ProviderApiCredentialUpsert,
superadmin=Depends(get_current_superadmin),
db: AsyncSession = Depends(get_db),
redis=Depends(get_redis),
):
"""Create or update shared provider API key (one key for all linked platforms)."""
if body.provider_id != provider_id:
from fastapi import HTTPException
raise HTTPException(400, "provider_id in path and body must match")
return await PlatformApiService(db, redis).upsert_provider_credential(
superadmin,
provider_id=provider_id,
api_key=body.api_key,
api_secret=body.api_secret,
is_enabled=body.is_enabled,
)
@router.post("/platform-apis/providers/{provider_id}/disable")
async def disable_provider_api(
provider_id: str,
superadmin=Depends(get_current_superadmin),
db: AsyncSession = Depends(get_db),
redis=Depends(get_redis),
):
"""Disable shared provider credential."""
scope_id = provider_scope_id(provider_id)
return await PlatformApiService(db, redis).disable_credential(superadmin, scope_id)
@router.delete("/platform-apis/providers/{provider_id}")
async def delete_provider_api(
provider_id: str,
superadmin=Depends(get_current_superadmin),
db: AsyncSession = Depends(get_db),
redis=Depends(get_redis),
):
"""Delete shared provider credential."""
scope_id = provider_scope_id(provider_id)
return await PlatformApiService(db, redis).delete_credential(superadmin, scope_id)