Skip to main content

Phase 01: Core Kernel & Interfaces

Objective: Set up the project foundation with clean hexagonal architecture. Define all protocol interfaces (ports) without any implementation. Principle: Test-Driven Development (TDD). Write tests first. Goal: 95% Coverage.


1. Project Initialization

  • Create project structure using uv:

    uv init framework-m
    cd framework-m
  • Setup pyproject.toml with dependencies

    • Add litestar[standard]>=2.0.0
    • Add sqlalchemy[asyncio]>=2.0.0 (ensure agnostic, test with sqlite)
    • Add pydantic>=2.0.0
    • Add pydantic-settings>=2.0.0
    • Add dependency-injector>=4.41.0 (powerful DI container)
    • Add asyncpg>=0.29.0 (PostgreSQL async driver)
    • Add redis>=5.0.0
    • Remove arq>=0.26.0
    • Add taskiq>=0.11.0, taskiq-nats>=0.4.0, nats-py>=2.0.0
    • Add dev dependencies: pytest, pytest-asyncio, mypy, ruff
  • Configure development tools

    • Setup mypy with strict mode in pyproject.toml
    • Configure ruff for linting and formatting
    • Add .gitignore for Python projects
  • Create directory structure

    src/framework_m/
    ├── core/
    │ ├── interfaces/ # Ports (protocols)
    │ └── domain/ # Domain models
    ├── adapters/ # Infrastructure (empty for now)
    ├── cli/ # CLI commands (empty for now)
    └── public/ # Built-in DocTypes (empty for now)

2. Define Port Interfaces

Key Principle: Ports are ONLY interfaces. No implementations. Use Python Protocol for structural typing.

2.1 Repository Protocol

  • Create tests/core/interfaces/test_repository.py
  • Define test for RepositoryProtocol interface compliance
  • Create src/framework_m/core/interfaces/repository.py
  • Define supporting models:
    • FilterSpec - Typed filter specification (field, operator, value)
    • OrderSpec - Sorting specification (field, direction)
    • PaginatedResult[T] - Result with items, total, limit, offset, has_more
  • Define RepositoryProtocol[T] (Generic):
    • async def get(self, id: UUID) -> T | None
    • async def save(self, entity: T, version: int | None = None) -> T (OCC support)
    • async def delete(self, id: UUID) -> None
    • async def exists(self, id: UUID) -> bool
    • async def count(self, filters: list[FilterSpec] | None = None) -> int
    • async def list(self, filters: list[FilterSpec] | None, order_by: list[OrderSpec] | None, limit: int, offset: int) -> PaginatedResult[T]
    • async def bulk_save(self, entities: list[T]) -> list[T]
  • Add type hints with Generic[T] and Protocol

2.2 Event Bus Protocol

  • Create src/framework_m/core/interfaces/event_bus.py
  • Define Event base model with id, timestamp, source, type, data
  • Define EventBusProtocol with methods:
    • async def connect(self) -> None (for NATS/Kafka)
    • async def disconnect(self) -> None
    • def is_connected(self) -> bool
    • async def publish(self, topic: str, event: Event) -> None
    • async def subscribe(self, topic: str, handler: Callable[[Event], Awaitable[None]]) -> str (returns subscription_id)
    • async def subscribe_pattern(self, pattern: str, handler: Callable) -> str (e.g., doc.*)
    • async def unsubscribe(self, subscription_id: str) -> None

2.3 Auth Context Protocol

  • Create src/framework_m/core/interfaces/auth_context.py
  • Define UserContext Pydantic model with fields:
    • id: str
    • email: str
    • roles: list[str]
    • tenants: list[str]
  • Define AuthContextProtocol with methods:
    • async def get_current_user() -> UserContext

2.4 Storage Protocol

  • Create src/framework_m/core/interfaces/storage.py
  • Define FileMetadata model with path, size, content_type, modified, etag
  • Define StorageProtocol with methods:
    • async def save_file(self, path: str, content: bytes, content_type: str | None = None) -> str
    • async def get_file(self, path: str) -> bytes
    • async def delete_file(self, path: str) -> None
    • async def list_files(self, prefix: str) -> list[str]
    • async def get_metadata(self, path: str) -> FileMetadata | None
    • async def get_url(self, path: str, expires: int = 3600) -> str (presigned URL for S3)
    • async def copy(self, src: str, dest: str) -> str
    • async def move(self, src: str, dest: str) -> str

2.5 Job Queue Protocol

  • Create src/framework_m/core/interfaces/job_queue.py
  • Define JobStatus enum: PENDING, RUNNING, SUCCESS, FAILED, CANCELLED
  • Define JobInfo model with id, name, status, enqueued_at, started_at, result, error
  • Define JobQueueProtocol with methods:
    • async def enqueue(self, job_name: str, **kwargs) -> str (returns job_id)
    • async def schedule(self, job_name: str, cron: str, **kwargs) -> str
    • async def cancel(self, job_id: str) -> bool
    • async def get_status(self, job_id: str) -> JobInfo | None
    • async def retry(self, job_id: str) -> str (returns new job_id)

2.6 Permission Protocol

  • Create src/framework_m/core/interfaces/permission.py
  • Define PermissionProtocol with methods:
    • async def has_permission(user: UserContext, doctype: str, action: str, doc_id: str | None) -> bool
    • async def get_permitted_filters(user: UserContext, doctype: str) -> dict
  • Add action types: "read", "write", "create", "delete", "submit"

2.7 Print Protocol

  • Create src/framework_m/core/interfaces/print.py
  • Define PrintProtocol with methods:
    • async def render(doc: BaseModel, template: str, format: str = "pdf") -> bytes

2.8 Cache Protocol

  • Create src/framework_m/core/interfaces/cache.py
  • Define CacheProtocol:
    • async def get(self, key: str) -> Any | None
    • async def set(self, key: str, value: Any, ttl: int | None = None) -> None
    • async def delete(self, key: str) -> None
    • async def exists(self, key: str) -> bool
    • async def get_many(self, keys: list[str]) -> dict[str, Any]
    • async def set_many(self, items: dict[str, Any], ttl: int | None = None) -> None
    • async def delete_pattern(self, pattern: str) -> int (returns count deleted)
    • async def ttl(self, key: str) -> int | None (remaining TTL)

2.9 Notification Protocol

  • Create src/framework_m/core/interfaces/notification.py
  • Define NotificationProtocol:
    • async def send_email(to: str, subject: str, body: str)
    • async def send_sms(to: str, body: str)

2.10 Search Protocol

  • Create src/framework_m/core/interfaces/search.py
  • Define SearchResult model with items, total, facets, highlights
  • Define SearchProtocol:
    • async def index(self, doctype: str, doc_id: str, doc: dict) -> None
    • async def delete_index(self, doctype: str, doc_id: str) -> None
    • async def search(self, doctype: str, query: str, filters: dict | None = None, limit: int = 20, offset: int = 0) -> SearchResult
    • async def reindex(self, doctype: str) -> int (returns count indexed)

2.11 I18n Protocol

  • Create src/framework_m/core/interfaces/i18n.py
  • Define I18nProtocol:
    • async def translate(text: str, locale: str) -> str
    • async def get_locale() -> str

3. Define Domain Layer

3.1 Base DocType

  • Create src/framework_m/core/domain/base_doctype.py
  • Define BaseDocType class inheriting from pydantic.BaseModel
  • Add standard fields:
    • name: Optional[str] (primary key, auto-generated if None)
    • creation: datetime
    • modified: datetime
    • modified_by: Optional[str]
    • owner: Optional[str]
  • Add Meta nested class for metadata:
    • layout: dict = {} (via get_layout())
    • permissions: dict = {} (via get_permissions())
  • Add class method get_doctype_name() -> str

3.2 Base Controller

  • Create src/framework_m/core/domain/base_controller.py
  • Define BaseController[T] generic class
  • Add lifecycle hook methods:
    • async def validate(self, context: Any = None) -> None
    • async def before_insert(self, context: Any = None) -> None
    • async def after_insert(self, context: Any = None) -> None
    • async def before_save(self, context: Any = None) -> None
    • async def after_save(self, context: Any = None) -> None
    • async def before_delete(self, context: Any = None) -> None
    • async def after_delete(self, context: Any = None) -> None
    • async def on_submit(self, context: Any = None) -> None
    • async def on_cancel(self, context: Any = None) -> None

3.3 Mixins

  • Create src/framework_m/core/domain/mixins.py
  • Define DocStatus enum:
    • DRAFT = 0
    • SUBMITTED = 1
    • CANCELLED = 2
  • Define SubmittableMixin class with:
    • docstatus: DocStatus = DocStatus.DRAFT
    • def is_submitted() -> bool
    • def is_cancelled() -> bool
    • def can_edit() -> bool (returns False if submitted)

4. Meta Registry

  • Create src/framework_m/core/registry.py
  • Implement MetaRegistry as singleton
  • Add storage dictionaries:
    • _doctypes: Dict[str, Type[BaseDocType]]
    • _controllers: Dict[str, Type[BaseController]]
  • Implement methods:
    • register_doctype(doctype_class, controller_class=None)
    • get_doctype(name: str) -> Type[BaseDocType]
    • get_controller(name: str) -> Type[BaseController] | None
    • list_doctypes() -> list[str]
    • discover_doctypes(package_name: str) (scans for BaseDocType subclasses)
    • Load Order: Follows installed_apps list.
    • Conflict Policy: Raise DuplicateDocTypeError if same name registered twice.

5. Port Implementation (Adapters)

5.1 Dependency Injection

  • Create tests/core/test_container.py

  • Write test for Container initialization and service resolution

  • Create src/framework_m/core/container.py

  • Implement Container class (using dependency_injector):

    from dependency_injector import containers, providers
  • Define Container class:

    class Container(containers.DeclarativeContainer):
    # Configuration
    config = providers.Configuration()

    # Core services (will be populated in later phases)
    # database = providers.Singleton(...)
    # repository_factory = providers.Factory(...)

5.2 Provider Types

  • Understand provider types:
    • Singleton - Single instance shared across app
    • Factory - New instance each time
    • Resource - For resources with lifecycle (db connections)
    • Callable - For functions
    • Dependency - For protocol injection

5.3 Configuration Provider

  • Setup configuration loading:
    config = providers.Configuration()
    config.from_pydantic(Settings()) # Pydantic settings
    config.from_env("APP", as_=str) # Environment variables

5.4 Wiring

  • Configure wiring for automatic injection:

    class Container(containers.DeclarativeContainer):
    wiring_config = containers.WiringConfiguration(
    modules=[
    "framework_m.adapters.web",
    "framework_m.adapters.db",
    ]
    )
  • Use @inject decorator in functions:

    from dependency_injector.wiring import inject, Provide

    @inject
    async def my_function(
    repo: RepositoryProtocol = Provide[Container.repository]
    ):
    # repo is automatically injected
    pass

5.5 Entrypoint Scanning for Overrides

  • Create override mechanism:

    • Scan framework_m.overrides entrypoints
    • Allow apps to override providers
    • Example:
      # In app's pyproject.toml
      [project.entry-points."framework_m.overrides"]
      repository = "my_app.custom:CustomRepository"
  • Implement override loader:

    def load_overrides(container: Container):
    for entry_point in entry_points(group="framework_m.overrides"):
    provider_name = entry_point.name
    override_class = entry_point.load()
    # Override the provider
    getattr(container, provider_name).override(override_class)

5.6 Testing Support

  • Use override for testing:

    # In tests
    container = Container()
    container.repository.override(MockRepository())
  • Reset overrides after tests:

    container.reset_singletons()
    container.unwire()

6. Testing Setup

  • Create tests/ directory structure

    tests/
    ├── unit/
    ├── integration/
    └── conftest.py
  • Setup conftest.py with fixtures:

    • pytest_asyncio configuration
    • Mock implementations of all protocols
  • Write initial tests:

    • Test MetaRegistry registration and retrieval
    • Test BaseDocType field validation
    • Test BaseController hook method existence

7. Documentation

  • Create README.md with:

    • Project overview
    • Installation instructions
    • Basic usage example
  • Create docs/ folder with:

    • architecture.md - Hexagonal architecture explanation
    • ports.md - List of all protocol interfaces (Repository, EventBus, Cache, Search, etc.)

Validation Checklist

Before moving to Phase 02, verify:

  • All protocol interfaces are defined with proper type hints
  • mypy --strict passes with no errors
  • No infrastructure dependencies in core/ (no imports of litestar, sqlalchemy, redis)
  • BaseDocType and BaseController are properly generic
  • MetaRegistry can register and retrieve DocTypes
  • All tests pass with pytest

Anti-Patterns to Avoid (Learning from Frappe)

Don't: Use global state like frappe.db or frappe.sessionDo: Use dependency injection to pass context

Don't: Import infrastructure libraries in domain layer ✅ Do: Define protocols and inject implementations

Don't: Use monkey patching for extensibility ✅ Do: Use DI and entrypoints for overrides

Don't: Mix business logic with database queries ✅ Do: Keep controllers focused on business rules, repositories handle data

Don't: Write code without tests ✅ Do: Follow TDD. 95% Coverage goal.