Merge pull request #13 from Athroniaeth/development

Development

Author: Athroniaeth

CHANGELOG.md

@@ -1,3 +1,23 @@
+## 0.9.0 (2025-11-26)
+
+### BREAKING CHANGE
+
+- please ensure that if you use sql repository to check how do you use delete_by_id method, this method now return entity and not boolean who define his existence
+
+### Fix
+
+- **cached**: use full API key hash to prevent key_id-only cache hits
+
+### Refactor
+
+- **svc**: replace entity parameter with direct args in create method
+- **svc**: create touch method for simplify code of service
+- **domain**: create ensure valid scopes domain method for regroup all verification of scopes
+
+### Perf
+
+- **sql**: remove double call to db for delete method
+
 ## 0.8.3 (2025-11-24)
 
 ### Fix

examples/example_sql_custom.py

@@ -1,111 +1,96 @@
+"""Example: Custom API Key with additional fields.
+
+This example demonstrates how to extend the default ApiKey entity and model
+with custom fields. Thanks to automatic introspection, you only need to:
+
+1. Create a custom dataclass extending ApiKey
+2. Create a custom SQLAlchemy model extending ApiKeyModelMixin
+3. Pass them to the repository - no need to override to_model/to_domain!
+
+The automatic mapping handles all common fields plus your custom ones.
+"""
+
+import asyncio
 import os
-from dataclasses import field, dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Optional, Type
+from typing import Optional
 
 from sqlalchemy import String
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
 from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
 
 from fastapi_api_key import ApiKeyService
-from fastapi_api_key.domain.entities import ApiKey as OldApiKey
+from fastapi_api_key.domain.entities import ApiKey
 from fastapi_api_key.hasher.argon2 import Argon2ApiKeyHasher
 from fastapi_api_key.repositories.sql import (
     SqlAlchemyApiKeyRepository,
     ApiKeyModelMixin,
 )
-import asyncio
 
 
+# 1. Define your custom SQLAlchemy Base
 class Base(DeclarativeBase): ...
 
 
+# 2. Create a custom domain entity with additional fields
 @dataclass
-class ApiKey(OldApiKey):
+class TenantApiKey(ApiKey):
+    """API Key with tenant isolation support."""
+
+    tenant_id: Optional[str] = field(default=None)
     notes: Optional[str] = field(default=None)
 
 
-class ApiKeyModel(Base, ApiKeyModelMixin):
+# 3. Create a custom SQLAlchemy model with matching columns
+class TenantApiKeyModel(Base, ApiKeyModelMixin):
+    """SQLAlchemy model with tenant support."""
+
+    tenant_id: Mapped[Optional[str]] = mapped_column(
+        String(64),
+        nullable=True,
+        index=True,  # Index for efficient tenant queries
+    )
     notes: Mapped[Optional[str]] = mapped_column(
-        String(128),
+        String(512),
         nullable=True,
     )
 
 
-class ApiKeyRepository(SqlAlchemyApiKeyRepository[ApiKey, ApiKeyModel]):
-    def __init__(
-        self,
-        async_session: AsyncSession,
-        model_cls: Type[ApiKeyModel] = ApiKeyModel,
-        domain_cls: Type[ApiKey] = ApiKey,
-    ) -> None:
-        super().__init__(
-            async_session=async_session,
-            model_cls=model_cls,
-            domain_cls=domain_cls,
-        )
-
-    @staticmethod
-    def to_model(
-        entity: ApiKey,
-        model_cls: Type[ApiKeyModel],
-        target: Optional[ApiKeyModel] = None,
-    ) -> ApiKeyModel:
-        if target is None:
-            return model_cls(
-                id_=entity.id_,
-                name=entity.name,
-                description=entity.description,
-                is_active=entity.is_active,
-                expires_at=entity.expires_at,
-                created_at=entity.created_at,
-                last_used_at=entity.last_used_at,
-                key_id=entity.key_id,
-                key_hash=entity.key_hash,
-                notes=entity.notes,
-            )
-
-        # Update existing model
-        target.name = entity.name
-        target.description = entity.description
-        target.is_active = entity.is_active
-        target.expires_at = entity.expires_at
-        target.last_used_at = entity.last_used_at
-        target.key_id = entity.key_id
-        target.key_hash = entity.key_hash  # type: ignore[invalid-assignment]
-        target.notes = entity.notes
-
-        return target
-
-    def to_domain(
-        self,
-        model: Optional[ApiKeyModel],
-        model_cls: Type[ApiKey],
-    ) -> Optional[ApiKey]:
-        if model is None:
-            return None
-
-        return model_cls(
-            id_=model.id_,
-            name=model.name,
-            description=model.description,
-            is_active=model.is_active,
-            expires_at=model.expires_at,
-            created_at=model.created_at,
-            last_used_at=model.last_used_at,
-            key_id=model.key_id,
-            key_hash=model.key_hash,
-            notes=model.notes,
-        )
+# 4. Create a factory function for your custom entity
+def tenant_api_key_factory(
+    key_id: str,
+    key_hash: str,
+    key_secret: str,
+    name: Optional[str] = None,
+    description: Optional[str] = None,
+    is_active: bool = True,
+    expires_at=None,
+    scopes=None,
+    tenant_id: Optional[str] = None,
+    notes: Optional[str] = None,
+    **kwargs,
+) -> TenantApiKey:
+    """Factory for creating TenantApiKey entities."""
+    return TenantApiKey(
+        key_id=key_id,
+        key_hash=key_hash,
+        _key_secret=key_secret,
+        name=name,
+        description=description,
+        is_active=is_active,
+        expires_at=expires_at,
+        scopes=scopes or [],
+        tenant_id=tenant_id,
+        notes=notes,
+    )
 
 
-# Set env var to override default pepper
-# Using a strong, unique pepper is crucial for security
-# Default pepper is insecure and should not be used in production
+# Configuration
 pepper = os.getenv("API_KEY_PEPPER")
 hasher = Argon2ApiKeyHasher(pepper=pepper)
 
-path = Path(__file__).parent / "db.sqlite3"
+path = Path(__file__).parent / "db_custom.sqlite3"
 database_url = os.environ.get("DATABASE_URL", f"sqlite+aiosqlite:///{path}")
 
 async_engine = create_async_engine(database_url, future=True)
@@ -117,21 +102,45 @@ def to_domain(
 
 
 async def main():
+    # Create tables
+    async with async_engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
     async with async_session_maker() as session:
-        repo = SqlAlchemyApiKeyRepository(session)
+        # Create repository with custom model and domain classes
+        # No need to override to_model/to_domain - automatic mapping handles it!
+        repo = SqlAlchemyApiKeyRepository(
+            async_session=session,
+            model_cls=TenantApiKeyModel,
+            domain_cls=TenantApiKey,
+        )
 
-        # Don't need to create Base and ApiKeyModel, the repository does it for you
-        await repo.ensure_table()
+        # Create service with custom factory
+        service = ApiKeyService(
+            repo=repo,
+            hasher=hasher,
+            entity_factory=tenant_api_key_factory,
+        )
 
-        service = ApiKeyService(repo=repo, hasher=hasher)
+        # Create an API key with custom fields
+        entity, secret = await service.create(
+            name="tenant-key",
+            description="API key for tenant operations",
+            scopes=["read", "write"],
+            tenant_id="tenant-123",
+            notes="Created for demo purposes",
+        )
 
-        # Entity have updated id after creation
-        entity, secret = await service.create(name="persistent")
-        print("Stored key", entity.id_, "secret", secret)
+        print(f"Created key for tenant: {entity.tenant_id}")
+        print(f"Notes: {entity.notes}")
+        print(f"Secret (store securely!): {secret}")
 
-        # Don't forget to commit the session to persist the key
-        # You can also use a transaction `async with session.begin():`
         await session.commit()
 
+        # Verify the key works
+        verified = await service.verify_key(secret)
+        print(f"\nVerified! Tenant: {verified.tenant_id}")
+
 
-asyncio.run(main())
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fastapi-api-key"
-version = "0.8.3"
+version = "0.9.0"
 description = "fastapi-api-key provides secure, production-ready API key management for FastAPI. It offers pluggable hashing strategies (Argon2, bcrypt, or custom), backend-agnostic persistence (SQLAlchemy, in-memory, or your own repository), and an optional cache layer (aiocache, Redis). Includes a Typer CLI and a FastAPI router for CRUD management of keys."
 readme = "README.md"
 authors = [

src/fastapi_api_key/api.py

@@ -20,6 +20,7 @@
 from fastapi.security import APIKeyHeader, HTTPBearer, HTTPAuthorizationCredentials
 from pydantic import BaseModel, Field
 
+from fastapi_api_key.repositories.base import ApiKeyFilter
 from fastapi_api_key.services.base import ApiKeyService
 from fastapi_api_key.domain.entities import ApiKey, ApiKeyEntity
 from fastapi_api_key.domain.errors import (
@@ -98,6 +99,50 @@ class DeletedResponse(BaseModel):
     status: Literal["deleted"] = "deleted"
 
 
+class ApiKeySearchIn(BaseModel):
+    """Search criteria for filtering API keys.
+
+    All criteria are optional. Only provided criteria are applied (AND logic).
+    """
+
+    is_active: Optional[bool] = Field(None, description="Filter by active status")
+    expires_before: Optional[datetime] = Field(None, description="Keys expiring before this date")
+    expires_after: Optional[datetime] = Field(None, description="Keys expiring after this date")
+    created_before: Optional[datetime] = Field(None, description="Keys created before this date")
+    created_after: Optional[datetime] = Field(None, description="Keys created after this date")
+    never_used: Optional[bool] = Field(None, description="True = never used keys, False = used keys")
+    scopes_contain_all: Optional[List[str]] = Field(None, description="Keys must have ALL these scopes")
+    scopes_contain_any: Optional[List[str]] = Field(None, description="Keys must have at least ONE of these scopes")
+    name_contains: Optional[str] = Field(None, description="Name contains this substring (case-insensitive)")
+    name_exact: Optional[str] = Field(None, description="Exact name match")
+
+    def to_filter(self, limit: int = 100, offset: int = 0) -> ApiKeyFilter:
+        """Convert to ApiKeyFilter with pagination."""
+        return ApiKeyFilter(
+            is_active=self.is_active,
+            expires_before=self.expires_before,
+            expires_after=self.expires_after,
+            created_before=self.created_before,
+            created_after=self.created_after,
+            never_used=self.never_used,
+            scopes_contain_all=self.scopes_contain_all,
+            scopes_contain_any=self.scopes_contain_any,
+            name_contains=self.name_contains,
+            name_exact=self.name_exact,
+            limit=limit,
+            offset=offset,
+        )
+
+
+class ApiKeySearchOut(BaseModel):
+    """Paginated search results."""
+
+    items: List[ApiKeyOut] = Field(description="List of matching API keys")
+    total: int = Field(description="Total number of matching keys (ignoring pagination)")
+    limit: int = Field(description="Page size used")
+    offset: int = Field(description="Offset used")
+
+
 def _to_out(entity: ApiKey) -> ApiKeyOut:
     """Map an `ApiKey` entity to the public `ApiKeyOut` schema."""
     return ApiKeyOut(
@@ -178,6 +223,40 @@ async def list_api_keys(
         items = await svc.list(offset=offset, limit=limit)
         return [_to_out(e) for e in items]
 
+    @router.post(
+        path="/search",
+        response_model=ApiKeySearchOut,
+        status_code=status.HTTP_200_OK,
+        summary="Search API keys with filters",
+    )
+    async def search_api_keys(
+        payload: ApiKeySearchIn,
+        svc: ApiKeyService = Depends(depends_svc_api_keys),
+        offset: Annotated[int, Query(ge=0, description="Items to skip")] = 0,
+        limit: Annotated[int, Query(gt=0, le=100, description="Page size")] = 50,
+    ) -> ApiKeySearchOut:
+        """Search API keys with advanced filtering criteria.
+
+        Args:
+            payload: Search criteria (all optional, AND logic).
+            svc: Injected `ApiKeyService`.
+            offset: Number of items to skip.
+            limit: Max number of items to return.
+
+        Returns:
+            Paginated search results with total count.
+        """
+        filter = payload.to_filter(limit=limit, offset=offset)
+        items = await svc.find(filter)
+        total = await svc.count(filter)
+
+        return ApiKeySearchOut(
+            items=[_to_out(e) for e in items],
+            total=total,
+            limit=limit,
+            offset=offset,
+        )
+
     @router.get(
         "/{api_key_id}",
         response_model=ApiKeyOut,