Spaces:
Running
Running
| """ | |
| Auth Models β Sessions, refresh tokens, and password reset tokens. | |
| """ | |
| import uuid | |
| from datetime import datetime | |
| from typing import Optional | |
| from sqlalchemy import String, Boolean, Text, Index, ForeignKey, DateTime | |
| from sqlalchemy.dialects.postgresql import UUID | |
| from sqlalchemy.orm import Mapped, mapped_column, relationship | |
| from app.models.base import Base, TimestampMixin, UUIDPrimaryKeyMixin | |
| class UserSession(UUIDPrimaryKeyMixin, TimestampMixin, Base): | |
| """ | |
| Active user sessions β tracks where a user is logged in. | |
| Enables 'logout from all devices' and session management. | |
| """ | |
| __tablename__ = "user_sessions" | |
| __table_args__ = ( | |
| Index("ix_user_sessions_user_id", "user_id"), | |
| Index("ix_user_sessions_is_active", "is_active"), | |
| ) | |
| user_id: Mapped[uuid.UUID] = mapped_column( | |
| UUID(as_uuid=True), | |
| ForeignKey("users.id", ondelete="CASCADE"), | |
| nullable=False, | |
| ) | |
| ip_address: Mapped[Optional[str]] = mapped_column(String(45), nullable=True) | |
| user_agent: Mapped[Optional[str]] = mapped_column(Text, nullable=True) | |
| device_info: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) | |
| is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False) | |
| last_activity_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) | |
| expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) | |
| # Relationships | |
| user: Mapped["User"] = relationship(back_populates="sessions") | |
| def __repr__(self) -> str: | |
| return f"<UserSession id={self.id} user_id={self.user_id}>" | |
| class RefreshToken(UUIDPrimaryKeyMixin, TimestampMixin, Base): | |
| """ | |
| Refresh token store β supports token rotation and family tracking. | |
| Token Family: When a refresh token is used, a new one is issued in the | |
| same family. If a revoked token is used again (replay attack), the entire | |
| family is revoked. | |
| """ | |
| __tablename__ = "refresh_tokens" | |
| __table_args__ = ( | |
| Index("ix_refresh_tokens_token_hash", "token_hash", unique=True), | |
| Index("ix_refresh_tokens_user_id", "user_id"), | |
| Index("ix_refresh_tokens_family_id", "family_id"), | |
| ) | |
| user_id: Mapped[uuid.UUID] = mapped_column( | |
| UUID(as_uuid=True), | |
| ForeignKey("users.id", ondelete="CASCADE"), | |
| nullable=False, | |
| ) | |
| session_id: Mapped[uuid.UUID] = mapped_column( | |
| UUID(as_uuid=True), | |
| ForeignKey("user_sessions.id", ondelete="CASCADE"), | |
| nullable=False, | |
| ) | |
| token_hash: Mapped[str] = mapped_column( | |
| String(255), unique=True, nullable=False | |
| ) | |
| family_id: Mapped[uuid.UUID] = mapped_column( | |
| UUID(as_uuid=True), default=uuid.uuid4, nullable=False | |
| ) | |
| is_revoked: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) | |
| revoked_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) | |
| revoked_reason: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) | |
| expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) | |
| def __repr__(self) -> str: | |
| return f"<RefreshToken id={self.id} user_id={self.user_id} revoked={self.is_revoked}>" | |
| class PasswordResetToken(UUIDPrimaryKeyMixin, TimestampMixin, Base): | |
| """Password reset token β single use, time-limited.""" | |
| __tablename__ = "password_reset_tokens" | |
| __table_args__ = ( | |
| Index("ix_password_reset_tokens_token_hash", "token_hash", unique=True), | |
| Index("ix_password_reset_tokens_user_id", "user_id"), | |
| ) | |
| user_id: Mapped[uuid.UUID] = mapped_column( | |
| UUID(as_uuid=True), | |
| ForeignKey("users.id", ondelete="CASCADE"), | |
| nullable=False, | |
| ) | |
| token_hash: Mapped[str] = mapped_column( | |
| String(255), unique=True, nullable=False | |
| ) | |
| is_used: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) | |
| used_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) | |
| expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) | |
| def __repr__(self) -> str: | |
| return f"<PasswordResetToken id={self.id} user_id={self.user_id}>" | |
| # Forward reference | |
| from app.models.user import User # noqa: E402 | |