Unified_Authentication_Platform/backend/app/services/hydra_service.py

import ory_hydra_client
from ory_hydra_client.api import o_auth2_api
from ory_hydra_client.models.accept_o_auth2_login_request import AcceptOAuth2LoginRequest
from ory_hydra_client.models.reject_o_auth2_request import RejectOAuth2Request
from ory_hydra_client.models.accept_o_auth2_consent_request import AcceptOAuth2ConsentRequest
from ory_hydra_client.models.o_auth2_consent_session import OAuth2ConsentSession

from app.core.hydra_config import hydra_settings

class HydraService:
    def __init__(self):
        configuration = ory_hydra_client.Configuration(
            host=hydra_settings.HYDRA_ADMIN_URL
        )
        self.api_client = ory_hydra_client.ApiClient(configuration)
        self.oauth2 = o_auth2_api.OAuth2Api(self.api_client)

    def get_login_request(self, challenge: str):
        return self.oauth2.get_o_auth2_login_request(challenge)

    def accept_login_request(self, challenge: str, subject: str):
        body = AcceptOAuth2LoginRequest(
            subject=subject,
            remember=True,
            remember_for=3600,
        )
        return self.oauth2.accept_o_auth2_login_request(challenge, accept_o_auth2_login_request=body)

    def reject_login_request(self, challenge: str, error: str, error_description: str):
        body = RejectOAuth2Request(
            error=error,
            error_description=error_description
        )
        return self.oauth2.reject_o_auth2_login_request(challenge, reject_o_auth2_request=body)

    def get_consent_request(self, challenge: str):
        return self.oauth2.get_o_auth2_consent_request(challenge)

    def accept_consent_request(self, challenge: str, grant_scope: list, id_token_claims: dict):
        body = AcceptOAuth2ConsentRequest(
            grant_scope=grant_scope,
            grant_access_token_audience=[],
            remember=True,
            remember_for=3600,
            session=OAuth2ConsentSession(
                id_token=id_token_claims
            )
        )
        return self.oauth2.accept_o_auth2_consent_request(challenge, accept_o_auth2_consent_request=body)
    
    def reject_consent_request(self, challenge: str, error: str, error_description: str):
        body = RejectOAuth2Request(
            error=error,
            error_description=error_description
        )
        return self.oauth2.reject_o_auth2_consent_request(challenge, reject_o_auth2_request=body)

hydra_service = HydraService()