app-base Developer Skill

This skill provides expert guidance for developing FastAPI applications using the app-base package. It focuses on the package's layered architecture, dependency injection, and hook-based customization patterns.

Core Philosophy & Architecture

The app-base package promotes a clean, layered architecture to build scalable and maintainable FastAPI applications. The key layers are:

•API/Router (api/): Handles HTTP requests, dependency injection, and calls UseCases.
•UseCase (usecases/): Orchestrates business logic, manages database transactions, and calls Services. Inherits from Base...UseCase classes.
•Service (services/): Implements the core business logic for a resource. Built with a mixin-based approach for CRUD operations and customized with hooks.
•Repository (repos/): Provides a generic data access layer for a specific SQLAlchemy model. Inherits from BaseRepository.
•Model (models/) & Schema (schemas/): Defines the database structure (SQLAlchemy) and data transfer objects (Pydantic).

Primary Workflow: Creating a New CRUD Endpoint

Follow these steps to create a new, complete CRUD endpoint for a resource (e.g., Book).

1. Define Model and Schemas

•

Model (models/book.py): Create the SQLAlchemy model. Use mixins from app_base.base.models.mixin for common fields like id, created_at, etc.

python

from app_base.base.models.mixin import Base, UUIDMixin, TimestampMixin
from sqlalchemy.orm import Mapped, mapped_column

class Book(Base, UUIDMixin, TimestampMixin):
    __tablename__ = "books"
    title: Mapped[str] = mapped_column(index=True)
    author: Mapped[str]

•

Schemas (schemas/book.py): Define Pydantic schemas for Create, Update, and Read.

python

from pydantic import BaseModel
from app_base.base.schemas.mixin import UUIDSchemaMixin, TimestampSchemaMixin

class BookBase(BaseModel):
    title: str
    author: str

class BookCreate(BookBase):
    pass

class BookUpdate(BaseModel):
    title: str | None = None
    author: str | None = None

class BookRead(BookBase, UUIDSchemaMixin, TimestampSchemaMixin):
    class Config:
        from_attributes = True

2. Create Repository

•

Repository (repos/book.py): Create a repository class that inherits from BaseRepository and links it to your model and schemas.

python

from app_base.base.repos.base import BaseRepository
from app.models.book import Book
from app.schemas.book import BookCreate, BookUpdate

class BookRepository(BaseRepository[Book, BookCreate, BookUpdate]):
    model = Book

3. Create Service & Apply Hooks

•

Service (services/book.py): This is where you compose CRUD functionality and add business logic using hooks. Inherit from the base service mixins and any desired hook mixins.

python

from app.repos.book import BookRepository
from app_base.base.services.base import (
    BaseCreateServiceMixin,
    BaseUpdateServiceMixin,
    BaseDeleteServiceMixin,
    BaseGetServiceMixin,
    BaseGetMultiServiceMixin,
)
# Import desired hooks
from app_base.base.services.exists_check_hook import ExistsCheckHooksMixin
from app_base.base.services.user_aware_hook import UserAwareHooksMixin, UserContextKwargs

class BookService(
    BaseCreateServiceMixin,
    BaseUpdateServiceMixin,
    BaseDeleteServiceMixin,
    BaseGetServiceMixin,
    BaseGetMultiServiceMixin,
    ExistsCheckHooksMixin, # Ensures book exists on update/delete
    UserAwareHooksMixin,  # Adds created_by/updated_by fields
):
    repo = BookRepository()
    context_model = UserContextKwargs # Specify context requirements

4. Create UseCases

•

UseCases (usecases/book.py): Wrap each service operation in a UseCase. This handles transactions and decouples the API layer from the service layer.

python

from app_base.base.usecases.crud import (
    BaseCreateUseCase,
    BaseGetUseCase,
    # ... import other use case bases
)
from app.services.book import BookService

service = BookService()

class CreateBookUseCase(BaseCreateUseCase):
    def __init__(self):
        super().__init__(service=service)

class GetBookUseCase(BaseGetUseCase):
    def __init__(self):
        super().__init__(service=service)

# ... define other use cases (Update, Delete, GetMulti)

5. Create API Router

•

Router (api/v1/books.py): Create the FastAPI router. Inject the UseCases and define the endpoints.

python

from fastapi import APIRouter, Depends
from app.usecases.book import CreateBookUseCase
# ... import other use cases

router = APIRouter()

@router.post("/")
async def create_book(
    # DI for use case, schemas, and context
):
    # return await CreateBookUseCase().execute(...)
    pass

# ... define other endpoints

Using Service Hooks

Hooks are the primary way to add business logic. Simply add the mixin to your service class.

•UserAwareHooksMixin: Automatically adds created_by and updated_by user IDs. Requires user_id: UUID to be in the context.
•ExistsCheckHooksMixin: Raises a NotFoundException if an object doesn't exist before an update or delete operation.
•
NestedResourceHooksMixin: For parent-child relationships.
- •You must implement the parent_repo property.
- •It automatically filters children by parent_id provided in the context.
- •It ensures a child belongs to the correct parent on get/update/delete.

•UniqueConstraintHooksMixin: Check for uniqueness before creating/updating. Implement the _unique_constraints async generator.

python

from sqlalchemy import and_

class MyService(UniqueConstraintHooksMixin, ...):
    async def _unique_constraints(self, obj_data, context):
        if obj_data.name:
            yield (
                and_(
                    self.repo.model.name == obj_data.name,
                    self.repo.model.parent_id == context["parent_id"]
                ),
                "Name must be unique within the parent."
            )

Using Adapters & AI

File Storage (S3, Local)

•The file storage client is initialized on application startup and can be accessed via a dependency.
•Dependency: app_base.adapter.file_storage.deps.get_file_storage_provider
•Interface: app_base.adapter.file_storage.interface.FileStorageClient (provides upload_file, download_file, etc.)

Vector Store (Qdrant)

•Use the VectorStoreFactory to get VectorStore instances for different collections.
•Dependency: app_base.adapter.vector_store.deps.get_vector_store_factory

•Usage:

python

def my_endpoint(
    vector_store_factory: Annotated[VectorStoreFactory, Depends(get_vector_store_factory)]
):
    vector_store = vector_store_factory.get_vector_store(
        collection_name="my_documents",
        model_name="text-embedding-ada-002"
    )
    # ... use vector_store

AI Model Factory (LangChain)

•A singleton AIModelFactory loads model configurations from catalog.yml.

•Usage:

python

from app_base.ai import AIModelFactory

factory = AIModelFactory()
llm = factory.get_llm("gpt-4-turbo")
embedding_model = factory.get_embedding("text-embedding-3-small")

Managing Application Configuration (Environment Variables)

The app-base project leverages Pydantic's BaseSettings for managing application configuration through environment variables. Configuration for different components (e.g., authentication, file storage, event broker) is defined in separate settings classes within the src/app_base/config/ directory.

To dynamically inspect the environment variables required for a specific configuration type, use the bundled get_env_spec.py script. This script will list the environment variables, their types, and default values based on the Pydantic settings definitions.

Prerequisite: The app-base package must be installed in your Python environment (e.g., via pip install app-base) for this script to function correctly.

Usage:

bash

python skill/app-base-developer-skill/scripts/get_env_spec.py --type <config_type>

Replace <config_type> with one of the following:

•auth: For general application settings (e.g., AuthSettings).
•app: For application-wide settings (e.g., AppSettings).
•
file_storage: For File Storage settings (lists FS_PROVIDER).
- •file_storage_none: For File Storage when provider is 'none'.
- •file_storage_local: For File Storage when provider is 'local'.
- •file_storage_s3: For File Storage when provider is 's3'.
•
event_broker: For Event Broker settings (lists EVENT_BROKER_PROVIDER).
- •event_broker_none: For Event Broker when provider is 'none'.
- •event_broker_in_memory: For Event Broker when provider is 'in_memory'.
- •event_broker_rabbitmq: For Event Broker when provider is 'rabbitmq'.
- •event_broker_redis: For Event Broker when provider is 'redis'.
•
vector_db: For Vector DB settings (lists VECTOR_DB_PROVIDER).
- •vector_db_none: For Vector DB when provider is 'none'.
- •vector_db_qdrant: For Vector DB when provider is 'qdrant'.
- •vector_db_milvus: For Vector DB when provider is 'milvus'.

Example:

bash

python skill/app-base-developer-skill/scripts/get_env_spec.py --type file_storage_s3

This command will output the environment variables and their specifications needed for configuring S3 file storage.

code


## Available Resources

When performing tasks, refer to these key files to understand the underlying patterns and base implementations.

-   `README.md`: High-level overview of the package.
-   `src/app_base/base/repos/base.py`: The `BaseRepository` implementation.
-   `src/app_base/base/services/base.py`: Defines all base service mixins and their hook interfaces (`_context_*`, `_prepare_*`, `_post_*`).
-   `src/app_base/base/services/*_hook.py`: Implementations for all standard service hooks. Review these to see how to add custom logic.
-   `src/app_base/base/usecases/crud.py`: The base UseCase implementations that manage transactions.
-   `src/app_base/ai/models/factory.py`: The `AIModelFactory` for creating LLM and embedding models.
-   `src/app_base/config/`: Pydantic settings management for different modules.
-   `src/app_base/adapter/`: Interfaces and implementations for file storage and vector stores.