feat(api_keys): implement API key management with CRUD operations and permissions

app/db/crud/admin.py

@@ -12,7 +12,7 @@
     get_complete_period_start_for_filter,
     to_utc_for_filter,
 )
-from app.db.models import Admin, AdminNotificationReminder, AdminRole, AdminUsageLogs, NodeUserUsage, ReminderType, User
+from app.db.models import APIKey, Admin, AdminNotificationReminder, AdminRole, AdminUsageLogs, NodeUserUsage, ReminderType, User
 from app.models.admin import (
     AdminCreate,
     AdminDetails,
@@ -235,6 +235,7 @@ async def remove_admin(db: AsyncSession, dbadmin: Admin) -> None:
         db (AsyncSession): Database session.
         dbadmin (Admin): The admin object to be removed.
     """
+    await db.execute(delete(APIKey).where(APIKey.admin_id == dbadmin.id))
     await db.delete(dbadmin)
     await db.commit()
 
@@ -771,6 +772,7 @@ async def remove_admins(db: AsyncSession, admin_ids: list[int]) -> None:
         return
 
     await db.execute(update(User).where(User.admin_id.in_(admin_ids)).values(admin_id=None))
+    await db.execute(delete(APIKey).where(APIKey.admin_id.in_(admin_ids)))
     await db.execute(delete(AdminUsageLogs).where(AdminUsageLogs.admin_id.in_(admin_ids)))
     await db.execute(delete(Admin).where(Admin.id.in_(admin_ids)))
     await db.commit()

app/db/crud/api_key.py

@@ -0,0 +1,125 @@
+import uuid
+from datetime import datetime as dt, timezone as tz
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.db.models import Admin, AdminStatus, APIKey, APIKeyStatus
+from app.models.api_key import APIKeyCreate
+from app.utils.crypto import api_key_lookup_id, hash_api_key, verify_api_key
+
+
+async def create_api_key(
+    db: AsyncSession,
+    admin_id: int,
+    model: APIKeyCreate,
+) -> tuple[str, APIKey]:
+    raw_uuid = str(uuid.uuid4())
+    raw_key = f"pg_key_{raw_uuid}"
+    db_key = APIKey(
+        admin_id=admin_id,
+        permissions={} if model.inherit_permissions else model.permissions.model_dump(exclude_none=True),
+        inherit_permissions=model.inherit_permissions,
+        name=model.name,
+        note=model.note,
+        key_hash=hash_api_key(raw_key),
+        api_key_trimmed=f"pg_key_{raw_uuid[:3]}***{raw_uuid[-3:]}",
+        expire_date=model.expire_date,
+    )
+    db.add(db_key)
+    await db.flush()
+    await db.refresh(db_key)
+    return raw_key, db_key
+
+
+async def get_api_key_by_raw_key(db: AsyncSession, raw_api_key: str) -> APIKey | None:
+    lookup_id = api_key_lookup_id(raw_api_key)
+
+    stmt = (
+        select(APIKey)
+        .where(
+            APIKey.key_hash.startswith(f"v1${lookup_id}$"),
+            APIKey.status != APIKeyStatus.disabled,
+        )
+        .options(selectinload(APIKey.admin).selectinload(Admin.role))
+        .limit(1)
+    )
+    db_key = (await db.execute(stmt)).scalar_one_or_none()
+
+    if db_key is None or not verify_api_key(raw_api_key, db_key.key_hash):
+        return None
+    # Reject if the owning admin is disabled
+    if db_key.admin is not None and db_key.admin.status == AdminStatus.disabled:
+        return None
+    return db_key
+
+
+async def get_api_key_by_id(db: AsyncSession, key_id: int) -> APIKey | None:
+    stmt = select(APIKey).where(APIKey.id == key_id).options(selectinload(APIKey.admin))
+    return (await db.execute(stmt)).scalar_one_or_none()
+
+
+async def get_api_keys_by_ids(db: AsyncSession, key_ids: list[int]) -> list[APIKey]:
+    if not key_ids:
+        return []
+
+    stmt = select(APIKey).where(APIKey.id.in_(key_ids)).options(selectinload(APIKey.admin))
+    return list((await db.execute(stmt)).scalars().all())
+
+
+async def get_api_keys(
+    db: AsyncSession,
+    *,
+    admin_id: int | None,
+    offset: int,
+    limit: int,
+    key_id: int | None = None,
+    name: str | None = None,
+    status: APIKeyStatus | None = None,
+) -> tuple[list[APIKey], int]:
+    stmt = select(APIKey)
+    if admin_id is not None:
+        stmt = stmt.where(APIKey.admin_id == admin_id)
+    if key_id is not None:
+        stmt = stmt.where(APIKey.id == key_id)
+    if name is not None:
+        stmt = stmt.where(APIKey.name == name)
+    if status is not None:
+        if status == APIKeyStatus.active:
+            # active = stored status is active AND not past expire_date
+            stmt = stmt.where(APIKey.status == APIKeyStatus.active, ~APIKey.is_expired)
+        else:
+            # disabled = stored status is disabled (expire_date irrelevant)
+            stmt = stmt.where(APIKey.status == APIKeyStatus.disabled)
+
+    total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
+
+    stmt = stmt.order_by(APIKey.created_at.desc()).offset(offset).limit(limit)
+    rows = list((await db.execute(stmt)).scalars().all())
+    return rows, total
+
+
+async def delete_api_key(db: AsyncSession, db_key: APIKey) -> None:
+    await db.delete(db_key)
+    await db.flush()
+
+
+async def revoke_api_key(db: AsyncSession, db_key: APIKey) -> tuple[str, APIKey]:
+    raw_uuid = str(uuid.uuid4())
+    raw_key = f"pg_key_{raw_uuid}"
+    db_key.key_hash = hash_api_key(raw_key)
+    db_key.api_key_trimmed = f"pg_key_{raw_uuid[:3]}***{raw_uuid[-3:]}"
+    db_key.revoked_at = dt.now(tz.utc)
+    db_key.status = APIKeyStatus.active
+    await db.flush()
+    await db.refresh(db_key)
+    return raw_key, db_key
+
+
+async def update_api_key(db: AsyncSession, db_key: APIKey, update_data: dict) -> APIKey:
+    for key, value in update_data.items():
+        setattr(db_key, key, value)
+    await db.flush()
+    await db.refresh(db_key)
+    return db_key

app/db/migrations/versions/9aa99aaee80f_.py

@@ -0,0 +1,24 @@
+"""empty message
+
+Revision ID: 9aa99aaee80f
+Revises: c9b48df42f10, d4f8c1b2a9e3
+Create Date: 2026-06-19 10:09:22.055892
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '9aa99aaee80f'
+down_revision = ('c9b48df42f10', 'd4f8c1b2a9e3')
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

app/db/migrations/versions/b6c9d0e1f2a3_api_key_inherit_permissions.py

@@ -0,0 +1,66 @@
+"""api key inherit permissions
+
+Revision ID: b6c9d0e1f2a3
+Revises: 9aa99aaee80f
+Create Date: 2026-06-19 20:30:00.000000
+
+"""
+
+import json
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "b6c9d0e1f2a3"
+down_revision = "9aa99aaee80f"
+branch_labels = None
+depends_on = None
+
+
+def _normalize_permissions(value):
+    if value is None:
+        return {}
+    if isinstance(value, str):
+        try:
+            return json.loads(value)
+        except json.JSONDecodeError:
+            return {}
+    if isinstance(value, dict):
+        return value
+    return {}
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("api_keys", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("inherit_permissions", sa.Boolean(), nullable=False, server_default="1"))
+
+    conn = op.get_bind()
+    admin_roles = sa.table(
+        "admin_roles",
+        sa.column("id", sa.Integer),
+        sa.column("permissions", sa.JSON),
+    )
+
+    rows = conn.execute(sa.select(admin_roles.c.id, admin_roles.c.permissions)).fetchall()
+    for role_id, role_permissions in rows:
+        permissions = _normalize_permissions(role_permissions)
+        api_key_permissions = permissions.get("api_keys")
+        if not isinstance(api_key_permissions, dict):
+            continue
+
+        changed = False
+        for action in ("read", "read_simple", "update", "delete"):
+            if api_key_permissions.get(action) is True:
+                api_key_permissions[action] = {"scope": 2}
+                changed = True
+
+        if changed:
+            permissions["api_keys"] = api_key_permissions
+            conn.execute(admin_roles.update().where(admin_roles.c.id == role_id).values(permissions=permissions))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("api_keys", schema=None) as batch_op:
+        batch_op.drop_column("inherit_permissions")

app/db/migrations/versions/c9b48df42f10_add_api_keys_table.py

@@ -0,0 +1,111 @@
+"""add api keys table
+
+Revision ID: c9b48df42f10
+Revises: f9c69a49f544
+Create Date: 2026-05-25 00:00:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+import app.db.compiles_types
+import json
+
+
+# revision identifiers, used by Alembic.
+revision = "c9b48df42f10"
+down_revision = "f9c69a49f544"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "api_keys",
+        sa.Column("id", app.db.compiles_types.SqliteCompatibleBigInteger(), autoincrement=True, nullable=False),
+        sa.Column("admin_id", app.db.compiles_types.SqliteCompatibleBigInteger(), nullable=False),
+        sa.Column("name", sa.String(length=128), nullable=False),
+        sa.Column("note", sa.String(length=512), nullable=True),
+        sa.Column("key_hash", sa.String(length=128), nullable=False),
+        sa.Column("api_key_trimmed", sa.String(length=16), nullable=False),
+        sa.Column("permissions", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("expire_date", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "status",
+            sa.Enum("active", "disabled", name="apikeystatus"),
+            nullable=False,
+            server_default="active",
+        ),
+        sa.ForeignKeyConstraint(
+            ["admin_id"], ["admins.id"], name=op.f("fk_api_keys_admin_id_admins"), ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_api_keys")),
+        sa.UniqueConstraint("key_hash", name=op.f("uq_api_keys_key_hash")),
+        sa.UniqueConstraint("admin_id", "name", name=op.f("uq_api_keys_admin_id")),
+    )
+    with op.batch_alter_table("api_keys", schema=None) as batch_op:
+        batch_op.create_index(batch_op.f("ix_api_keys_admin_id"), ["admin_id"], unique=False)
+        batch_op.create_index(batch_op.f("ix_api_keys_created_at"), ["created_at"], unique=False)
+        batch_op.create_index(batch_op.f("ix_api_keys_expire_date"), ["expire_date"], unique=False)
+
+    # Update admin_roles permissions to include api_keys entry
+    OWNER_ADMIN_API_KEY_PERMS = {
+        "create": True,
+        "read": {"scope": 2},
+        "read_simple": {"scope": 2},
+        "update": {"scope": 2},
+        "delete": {"scope": 2},
+    }
+    OPERATOR_API_KEY_PERMS = {
+        "read": {"scope": 1},
+        "read_simple": {"scope": 1},
+        "update": {"scope": 1},
+        "delete": {"scope": 1},
+    }
+
+
+    def _normalize_permissions(value):
+        if value is None:
+            return {}
+        if isinstance(value, str):
+            try:
+                return json.loads(value)
+            except json.JSONDecodeError:
+                return {}
+        if isinstance(value, dict):
+            return value
+        return {}
+
+    conn = op.get_bind()
+    admin_roles = sa.table(
+        "admin_roles",
+        sa.column("id", sa.Integer),
+        sa.column("name", sa.String),
+        sa.column("permissions", sa.JSON),
+    )
+
+    rows = conn.execute(sa.select(admin_roles.c.id, admin_roles.c.name, admin_roles.c.permissions)).fetchall()
+    for role_id, role_name, role_permissions in rows:
+        permissions = _normalize_permissions(role_permissions)
+        if "api_keys" in permissions:
+            continue
+        if role_name in {"owner", "administrator"}:
+            permissions["api_keys"] = OWNER_ADMIN_API_KEY_PERMS
+        else:
+            permissions["api_keys"] = OPERATOR_API_KEY_PERMS
+        conn.execute(admin_roles.update().where(admin_roles.c.id == role_id).values(permissions=permissions))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("api_keys", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_api_keys_expire_date"))
+        batch_op.drop_index(batch_op.f("ix_api_keys_created_at"))
+        batch_op.drop_index(batch_op.f("ix_api_keys_admin_id"))
+
+    op.drop_table("api_keys")
+
+    conn = op.get_bind()
+    if conn.dialect.name == "postgresql":
+        op.execute("DROP TYPE IF EXISTS apikeystatus")