1  from __future__ import annotations
 2  
 3  from typing import Any
 4  from uuid import uuid4
 5  
 6  from sqlalchemy import Column, ForeignKey, String, Text
 7  from sqlalchemy.orm import Mapped, relationship
 8  
 9  from .user import User
10  from ..database import Base, db
11  
12  
13  class OAuthUserConnection(Base):
14      __tablename__ = "oauth_user_connection"
15  
16      id: Mapped[str] = Column(String(36), primary_key=True, unique=True)
17      user_id: Mapped[str] = Column(String(36), ForeignKey("user.id"))
18      user: User = relationship("User", back_populates="oauth_connections")
19      provider_id: Mapped[str] = Column(String(64))
20      remote_user_id: Mapped[str] = Column(Text)
21      display_name: Mapped[str | None] = Column(Text, nullable=True)
22  
23      @property
24      def serialize(self) -> dict[str, Any]:
25          return {"id": self.id, "provider_id": self.provider_id, "display_name": self.display_name}
26  
27      @staticmethod
28      async def create(
29          user_id: str, provider_id: str, remote_user_id: str, display_name: str | None
30      ) -> OAuthUserConnection:
31          row = OAuthUserConnection(
32              id=str(uuid4()),
33              user_id=user_id,
34              provider_id=provider_id,
35              remote_user_id=remote_user_id,
36              display_name=display_name,
37          )
38          await db.add(row)
39          return row