Build and Push / build (release) Successful in 1m16s
- Add PostgreSQL database with async SQLAlchemy - Add Alembic migrations for schema management - Add User, Role, UserPreferences, ApiKey models - Add auth endpoints: /auth/me, /auth/users, /auth/users/sync-from-authentik - Add token validation via Authentik userinfo endpoint - Add bulk user sync from Authentik admin API - Add database health check to diagnostics 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
99 lines
2.7 KiB
Python
99 lines
2.7 KiB
Python
"""
|
|
API Key Model
|
|
|
|
Provides API key authentication as fallback for OIDC.
|
|
Keys are tied to user accounts and inherit user permissions.
|
|
"""
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import String, ForeignKey, DateTime, func
|
|
from sqlalchemy.dialects.postgresql import UUID, ARRAY
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from src.db.database import Base
|
|
|
|
from typing import TYPE_CHECKING, List
|
|
|
|
if TYPE_CHECKING:
|
|
from src.db.models.user import User
|
|
|
|
|
|
class ApiKey(Base):
|
|
"""
|
|
API Key model for programmatic access
|
|
|
|
API keys provide an alternative to OIDC for:
|
|
- Local development without SSO
|
|
- Service-to-service communication
|
|
- Scripts and automation
|
|
|
|
Keys inherit the user's roles but can optionally
|
|
be restricted to a subset of scopes.
|
|
"""
|
|
|
|
__tablename__ = "api_keys"
|
|
|
|
id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
primary_key=True,
|
|
default=uuid.uuid4,
|
|
)
|
|
user_id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("users.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
name: Mapped[str] = mapped_column(
|
|
String(100),
|
|
nullable=False,
|
|
comment="Human-readable key name (e.g., 'Dev Laptop', 'CI/CD')",
|
|
)
|
|
key_hash: Mapped[str] = mapped_column(
|
|
String(255),
|
|
nullable=False,
|
|
comment="SHA-256 hash of the API key",
|
|
)
|
|
key_prefix: Mapped[str] = mapped_column(
|
|
String(8),
|
|
nullable=False,
|
|
comment="First 8 chars of key for identification (e.g., 'cak_abc1')",
|
|
)
|
|
scopes: Mapped[List[str] | None] = mapped_column(
|
|
ARRAY(String),
|
|
nullable=True,
|
|
comment="Optional scope restriction (subset of user roles)",
|
|
)
|
|
expires_at: Mapped[datetime | None] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=True,
|
|
comment="Optional expiration timestamp",
|
|
)
|
|
last_used_at: Mapped[datetime | None] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=True,
|
|
comment="Last time this key was used",
|
|
)
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
server_default=func.now(),
|
|
nullable=False,
|
|
)
|
|
|
|
# Relationships
|
|
user: Mapped["User"] = relationship(
|
|
"User",
|
|
back_populates="api_keys",
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<ApiKey {self.key_prefix}... ({self.name})>"
|
|
|
|
@property
|
|
def is_expired(self) -> bool:
|
|
"""Check if the API key has expired"""
|
|
if self.expires_at is None:
|
|
return False
|
|
return datetime.now(self.expires_at.tzinfo) > self.expires_at
|