Class UserIdentity

app/models.py:95–165

class UserIdentity(Base):
    __tablename__: str = "user_identity"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), index=True)
    provider: Mapped[str] = mapped_column(
        SQLAEnum("github", "google", name="identity_provider"),
        nullable=False,
        index=True,
    )
    provider_user_id: Mapped[str | None] = mapped_column(
        String(100), nullable=True, index=True
    )
    _access_token: Mapped[str | None] = mapped_column(
        "access_token", String(2048), nullable=True
    )
    _refresh_token: Mapped[str | None] = mapped_column(
        "refresh_token", String(2048), nullable=True
    )
    token_expires_at: Mapped[datetime | None] = mapped_column(nullable=True)
    password_hash: Mapped[str | None] = mapped_column(
        String(255), nullable=True
    )  # Only for password provider
    provider_metadata: Mapped[dict | None] = mapped_column(
        JSON, nullable=True
    )  # Store provider-specific data
    created_at: Mapped[datetime] = mapped_column(default=utc_now)
    updated_at: Mapped[datetime] = mapped_column(default=utc_now, onupdate=utc_now)

    # Relationships
    user: Mapped[User] = relationship(back_populates="identities")

    __table_args__ = (
        UniqueConstraint(
            "provider", "provider_user_id", name="uq_identity_provider_user"
        ),
    )

    @property
    def access_token(self) -> str | None:
        if self._access_token:
            fernet = get_fernet()
            return fernet.decrypt(self._access_token.encode()).decode()
        return None

    @access_token.setter
    def access_token(self, value: str | None):
        if value:
            fernet = get_fernet()
            self._access_token = fernet.encrypt(value.encode()).decode()
        else:
            self._access_token = None

    @property
    def refresh_token(self) -> str | None:
        if self._refresh_token:
            fernet = get_fernet()
            return fernet.decrypt(self._refresh_token.encode()).decode()