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.tomlwith 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
- Add
-
Configure development tools
- Setup
mypywith strict mode inpyproject.toml - Configure
rufffor linting and formatting - Add
.gitignorefor Python projects
- Setup
-
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
RepositoryProtocolinterface 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 withitems,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]andProtocol
2.2 Event Bus Protocol
- Create
src/framework_m/core/interfaces/event_bus.py - Define
Eventbase model withid,timestamp,source,type,data - Define
EventBusProtocolwith 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
UserContextPydantic model with fields:-
id: str -
email: str -
roles: list[str] -
tenants: list[str]
-
- Define
AuthContextProtocolwith methods:-
async def get_current_user() -> UserContext
-
2.4 Storage Protocol
- Create
src/framework_m/core/interfaces/storage.py - Define
FileMetadatamodel withpath,size,content_type,modified,etag - Define
StorageProtocolwith 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
JobStatusenum:PENDING,RUNNING,SUCCESS,FAILED,CANCELLED - Define
JobInfomodel withid,name,status,enqueued_at,started_at,result,error - Define
JobQueueProtocolwith 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
PermissionProtocolwith 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
PrintProtocolwith 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
SearchResultmodel withitems,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
BaseDocTypeclass inheriting frompydantic.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
Metanested class for metadata:-
layout: dict = {}(viaget_layout()) -
permissions: dict = {}(viaget_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
DocStatusenum:-
DRAFT = 0 -
SUBMITTED = 1 -
CANCELLED = 2
-
- Define
SubmittableMixinclass 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
MetaRegistryas 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_appslist. - Conflict Policy: Raise
DuplicateDocTypeErrorif 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
Containerclass (usingdependency_injector):from dependency_injector import containers, providers -
Define
Containerclass: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
@injectdecorator 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.overridesentrypoints - Allow apps to override providers
- Example:
# In app's pyproject.toml
[project.entry-points."framework_m.overrides"]
repository = "my_app.custom:CustomRepository"
- Scan
-
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
overridefor testing:# In tests
container = Container()
container.repository.override(MockRepository()) -
Reset overrides after tests:
container.reset_singletons()
container.unwire()
6. Testing Setup
-
Create
tests/directory structuretests/
├── unit/
├── integration/
└── conftest.py -
Setup
conftest.pywith fixtures:-
pytest_asyncioconfiguration - Mock implementations of all protocols
-
-
Write initial tests:
- Test
MetaRegistryregistration and retrieval - Test
BaseDocTypefield validation - Test
BaseControllerhook method existence
- Test
7. Documentation
-
Create
README.mdwith:- 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 --strictpasses with no errors - No infrastructure dependencies in
core/(no imports of litestar, sqlalchemy, redis) -
BaseDocTypeandBaseControllerare properly generic -
MetaRegistrycan 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.session
✅ Do: 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.