import json import time from authlib.integrations.sqla_oauth2 import OAuth2ClientMixin as BaseOAuth2ClientMixin, OAuth2AuthorizationCodeMixin as BaseOAuth2AuthorizationCodeMixin, OAuth2TokenMixin as BaseOAuth2TokenMixin from sqlalchemy.sql.sqltypes import Text from sqlmodel import Field class OAuth2ClientMixin(BaseOAuth2ClientMixin): client_id: str = Field(max_length=48, index=True, nullable=True) client_secret: str = Field(max_length=120, nullable=True) client_id_issued_at: int = Field(nullable=False, default=0) client_secret_expires_at: int = Field(nullable=False, default=0) client_metadata_field: str = Field(sa_type=Text, sa_column_kwargs={ 'name': 'client_metadata' }, nullable=True) _client_metadata = None @property def client_metadata(self) -> dict: if self.client_metadata_field: return json.loads(self.client_metadata_field) return {} def set_client_metadata(self, metadata: dict): self.client_metadata_field = json.dumps(metadata) class OAuth2AuthorizationCodeMixin(BaseOAuth2AuthorizationCodeMixin): code: str = Field(max_length=120, unique=True, nullable=False) client_id: str = Field(max_length=48, nullable=True) redirect_uri: str = Field(sa_type=Text, default="", nullable=True) response_type: str = Field(sa_type=Text, default="", nullable=True) scope: str = Field(sa_type=Text, default="", nullable=True) nonce: str = Field(sa_type=Text, nullable=True) auth_time: int = Field(nullable=False, default_factory=lambda: int(time.time())) acr: str = Field(sa_type=Text, nullable=True) amr: str = Field(sa_type=Text, nullable=True) code_challenge: str = Field(sa_type=Text, nullable=True) code_challenge_method: str = Field(max_length=48, nullable=True) class OAuth2TokenMixin(BaseOAuth2TokenMixin): client_id: str = Field(max_length=48, nullable=True) token_type: str = Field(max_length=40, nullable=True) access_token: str = Field(max_length=255, unique=True, nullable=False) refresh_token: str = Field(max_length=255, index=True, nullable=True) scope: str = Field(sa_type=Text, default="", nullable=True) issued_at: int = Field(nullable=False, default_factory=lambda: int(time.time())) access_token_revoked_at: int = Field(nullable=False, default=0) refresh_token_revoked_at: int = Field(nullable=False, default=0) expires_in: int = Field(nullable=False, default=0)