Repository: Significant-Gravitas/Auto-GPT
File: rnd/autogpt_server/autogpt_server/server/integrations.py (Python)

Code before:
    refresh_token: Optional[str]
    access_token_expires_at: Optional[int]
    refresh_token_expires_at: Optional[int]
    scopes: list[str]
    metadata: dict


@integrations_api_router.get(
    "/{provider}/credentials/{cred_id}", response_model=CredentialsDetailResponse
)
async def get_credential(
    provider: Annotated[str, Path(title="The provider to retrieve credentials for")],
    cred_id: Annotated[str, Path(title="The ID of the credentials to retrieve")],
    user_id: Annotated[str, Depends(get_user_id)],
    store: Annotated[SupabaseIntegrationCredentialsStore, Depends(get_store)],
) -> CredentialsDetailResponse:
    credential = store.get_creds_by_id(user_id, cred_id)
    if not credential:
        raise HTTPException(status_code=404, detail="Credentials not found")
    if credential.provider != provider:
        raise HTTPException(
            status_code=400, detail="Credentials do not match the specified provider"
        )

    if isinstance(credential, OAuth2Credentials):
        return CredentialsDetailResponse(
            id=credential.id,
            provider=credential.provider,
            title=credential.title,
            type=credential.type,
            access_token=credential.access_token.get_secret_value(),
            refresh_token=(
                credential.refresh_token.get_secret_value()
                if credential.refresh_token
                else None
            ),
            access_token_expires_at=credential.access_token_expires_at,
            refresh_token_expires_at=credential.refresh_token_expires_at,
            scopes=credential.scopes,
            metadata=credential.metadata,
        )
    elif isinstance(credential, APIKeyCredentials):
        return CredentialsDetailResponse(
            id=credential.id,
            provider=credential.provider,
            title=credential.title,
            type=credential.type,
            access_token=credential.api_key.get_secret_value(),
            refresh_token=None,
            access_token_expires_at=credential.expires_at,
            refresh_token_expires_at=None,

Review comment (refactor):
Should I split this into two models, one for each type?

Code after:
        )
        for cred in credentials
    ]


@integrations_api_router.get("/{provider}/credentials/{cred_id}")
async def get_credential(
    provider: Annotated[str, Path(title="The provider to retrieve credentials for")],
    cred_id: Annotated[str, Path(title="The ID of the credentials to retrieve")],
    user_id: Annotated[str, Depends(get_user_id)],
    store: Annotated[SupabaseIntegrationCredentialsStore, Depends(get_store)],
) -> Credentials:
    credential = store.get_creds_by_id(user_id, cred_id)
    if not credential:
        raise HTTPException(status_code=404, detail="Credentials not found")
    if credential.provider != provider:
        raise HTTPException(
            status_code=404, detail="Credentials do not match the specified provider"
        )
    return credential


# -------- UTILITIES --------- #


def _get_provider_oauth_handler(req: Request, provider_name: str) -> BaseOAuthHandler:
    if provider_name not in HANDLERS_BY_NAME:
        raise HTTPException(
            status_code=404, detail=f"Unknown provider '{provider_name}'"
        )

    client_id = getattr(settings.secrets, f"{provider_name}_client_id")
    client_secret = getattr(settings.secrets, f"{provider_name}_client_secret")
    if not (client_id and client_secret):
        raise HTTPException(
            status_code=501,
            detail=f"Integration with provider '{provider_name}' is not configured",
        )

    handler_class = HANDLERS_BY_NAME[provider_name]
    return handler_class(
        client_id=client_id,
        client_secret=client_secret,
        redirect_uri=str(req.url_for("callback", provider=provider_name)),
    )