Phase 03: API Layer & Authorization
Objective: Expose DocTypes via HTTP with auto-generated CRUD endpoints, implement permission system with row-level security, and add RPC support.
1. Litestar Application Setup
-
Create
src/framework_m/adapters/web/app.py -
Initialize Litestar app:
- Configure CORS for development
- Add exception handlers
- Configure OpenAPI documentation
- Integrate with
dependency-injectorContainer (wire modules) - Middleware Discovery:
- Scan
MiddlewareRegistryfor registered App Middlewares. - Allow Apps to inject ASGI Middleware (e.g.
RateLimit,cors,Compression). - APM & Tracing: Support OpenTelemetry, DataDog, Sentry via standard ASGI middleware.
- Support ordering (priority).
- Scan
-
Add application lifecycle:
-
on_startup: Initialize Container, database, discover DocTypes -
on_shutdown: Close database connections, unwire Container
-
2. Authentication Middleware
-
Create
src/framework_m/adapters/web/middleware.py -
Implement
AuthMiddleware:- Extract
x-user-idheader - Extract
x-rolesheader (comma-separated) - Extract
x-tenantsheader (optional, comma-separated) - Create
UserContextobject - Store in
request.state.user - Return 401 if headers missing (configurable)
- Extract
-
Add middleware to Litestar app (available via
create_auth_middleware())
3. Permission System
3.1 Permission Protocol Implementation
- Create
src/framework_m/adapters/auth/rbac_permission.py - Implement
RbacPermissionAdapter:- Implement
evaluate(request: PolicyEvaluateRequest) -> PolicyEvaluateResult - Read permissions from
DocType.Meta.permissions - Match
request.principal_attributes["roles"]against allowed roles -
CustomPermissionDocType created (integration pending) - Return
PolicyEvaluateResult(authorized=True/False, decision_source=DecisionSource.RBAC)
- Implement
3.2 Permission Configuration
- Define permission structure in DocType:
class Meta:
# Access Control Flags
requires_auth: bool = True # Does user need to be logged in?
apply_rls: bool = True # Is Row-Level Security applied?
# RBAC Permissions (only used by RbacPermissionAdapter)
permissions = {
"read": ["Employee", "Manager"],
"write": ["Manager"],
"create": ["Manager"],
"delete": ["Admin"],
"submit": ["Manager"]
}
Access Control Combinations:
requires_auth | apply_rls | Use Case |
|---|---|---|
True | True | Default. Alice only sees her own invoices. |
True | False | Lookup tables. All users see all countries. |
False | False | Public data. App config, announcements. |
False | True | Public profiles. Only owner's data exposed. |
- Implement
has_permission():- Check
requires_auth— if False and no user, allow read - Check if user has required role
- Check database overrides via
PermissionLookupService(CustomPermission DocType) - For object-level: check ownership or custom rules
- Return boolean
- Check
3.3 System Context (Elevated Operations)
[!IMPORTANT] No
ignore_permissionsflag. For system operations, use explicitSystemContext.
For background jobs, migrations, or service-to-service calls that need elevated access:
- Create
src/framework_m/core/system_context.py - Implement
SystemContext:@asynccontextmanager
async def system_context():
"""Context manager for system-level operations.
All operations are logged as 'system' principal.
Use only in background jobs, migrations, webhooks.
"""
yield SystemPrincipal(
id="system",
roles=["System"],
is_system=True,
) - Usage in background jobs:
async def sync_external_data():
async with system_context() as ctx:
# Runs with system principal, bypasses RLS
await repo.save(ctx.session, data) - Audit: All system operations logged with
principal="system"
3.4 Row-Level Security (RLS)
-
Implement
get_permitted_filters():- For standard users: add
WHERE owner = :user_id - For admins: no filter (see all)
- For custom rules: via
Meta.rls_field(completed by Section 3.5 Team-Based Access) - Return dict of SQLAlchemy filters
- For standard users: add
-
Update
GenericRepository.list():-
list_for_user()method with RLS filtering -
apply_rls_filters()helper available - Merge with user-provided filters
- Apply to SQL query
-
-
Update
GenericRepository.get():-
get_for_user()method with permission check - Raise
PermissionDeniedErrorif denied
-
3.5 Team-Based Access (Indie)
For team/department-level access without full ReBAC:
- Add
Meta.rls_fieldoption:class Invoice(BaseDocType):
owner: str
team: str # e.g., "sales"
class Meta:
apply_rls = True
rls_field = "team" # RLS: WHERE team IN :user_teams - Update
get_permitted_filters():- If
rls_field = "owner"(default):WHERE owner = :user_id - If
rls_field = "team":WHERE team IN :user_teams - Support custom fields
- If
[!NOTE] Where does
user_teamscome from?
- From
PolicyEvaluateRequest.principal_attributes["teams"]- The auth layer (middleware) populates this from headers, JWT claims, or external IAM
- Framework M core does NOT assume a built-in User model
- Indie: Phase 06 adds optional
User+UserAttributeDocTypes for self-contained apps
3.6 Explicit Sharing (DocumentShare)
For sharing specific documents with specific users/roles:
- Create built-in
DocumentShareDocType:class DocumentShare(BaseDocType):
doctype_name: str # e.g., "Invoice"
doc_id: str # e.g., "INV-001"
shared_with: str # User ID or Role name
share_type: ShareType # USER or ROLE
granted_permissions: list[str] # ["read"] or ["read", "write"]
class Meta:
requires_auth = True
apply_rls = True # Users only see shares they created - Update
get_permitted_filters()to include shares viaget_rls_filters_with_shares():# Returns filters including shared doc IDs:
# {"owner": "user-123", "_shared_doc_ids": ["INV-001", "INV-002"]}
filters = await get_rls_filters_with_shares(user, "Invoice", shares, "read") - Add share management API:
-
POST /api/v1/share— Create share -
DELETE /api/v1/share/{id}— Remove share -
GET /api/v1/{doctype}/{id}/shares— List shares for a doc
-
[!NOTE] Enterprise Alternative: For complex sharing graphs, swap to SpiceDB adapter. Relationships like "team member", "org hierarchy", "inherited access" are better in ReBAC.
3.7 Indie Mode: Permission Conveniences
[!TIP] 0-Cliff Principle: These helpers internally construct
PolicyEvaluateRequest. Indie devs get simplicity; enterprise devs can use the full API when needed.
@requires_permission Decorator
- Create
src/framework_m/core/decorators.py(if not exists) - Implement
@requires_permission(action: PermissionAction):@requires_permission(PermissionAction.WRITE)
async def update_invoice(self, invoice_id: str, data: InvoiceUpdate):
# Permission already checked, just do business logic
...- Internally builds
PolicyEvaluateRequestfromself.user,self.doctype_name - Calls
permission.evaluate(request) - Raises
PermissionDeniedErrorif not authorized - Works on Controller methods
- Internally builds
check_permission() Helper
- Add to
BaseController:async def check_permission(self, action: str, doc_id: str | None = None) -> bool:
"""Simple helper that builds PolicyEvaluateRequest internally."""
# Returns bool - use require_permission() to raise on failure
require_permission() Helper
- Add to
BaseController:async def require_permission(self, action: str, doc_id: str | None = None) -> None:
"""Raises PermissionDeniedError if not authorized."""
Auto-Permission in CRUD Routes
- Verify: All auto-generated CRUD routes (Section 4) automatically call
permission.evaluate():-
POST /api/v1/{doctype}→ checksCREATEpermission (verified via test) -
GET /api/v1/{doctype}/{id}→ checksREADpermission (verified via test) -
PUT /api/v1/{doctype}/{id}→ checksWRITEpermission (verified via test) -
DELETE /api/v1/{doctype}/{id}→ checksDELETEpermission (verified via test)
-
- No manual code required for indie devs using auto-CRUD (verified via test)
4. Auto-CRUD Router
4.1 Meta Router Implementation
- Create
src/framework_m/adapters/web/meta_router.py - Implement
create_crud_routes(doctype_class: type[BaseDocType]):- Check Opt-in: Verify
DocType.Meta.api_resource is True. Skip if False. - Generate 5 routes per DocType
- Inject repository and permission service
- Parse filters from JSON string via
_parse_filters()
- Check Opt-in: Verify
4.2 List Endpoint
- Create
GET /api/v1/{doctype}:- Accept query parameters:
limit,offset,filters,order_by - Parse filters from JSON string
- Call repository
list_entities()with RLS filters viaapply_rls_filters() - Return paginated response with metadata:
{
"data": [...],
"total": 100,
"limit": 20,
"offset": 0
}
- Accept query parameters:
4.3 Create Endpoint
- Create
POST /api/v1/{doctype}:- Validate request body against Pydantic model
- Check
CREATEpermission via_check_permission() - Set
ownerto current user (data["owner"] = user_id) - Call repository
save() - Return 201 with created document
4.4 Read Endpoint
- Create
GET /api/v1/{doctype}/{id}:- Call repository
get(id) - Check
READpermission - Return 404 if not found (via
NotFoundException) - Return 403 if permission denied
- Return document
- Call repository
4.5 Update Endpoint
- Create
PUT /api/v1/{doctype}/{id}:- Load existing document via
repo.get() - Check
WRITEpermission - Check if submitted (deny if immutable) via
_check_submitted() - Merge changes via
entity.model_copy(update=data) - Call repository
save() - Return updated document
- Load existing document via
4.6 Delete Endpoint
- Create
DELETE /api/v1/{doctype}/{id}:- Load document
- Check
DELETEpermission - Check if submitted (deny if immutable) via
_check_submitted() - Call repository
delete() - Return 204 No Content
4.7 Router Registration
- Implement
create_meta_router():- Get all DocTypes from MetaRegistry
- For each DocType with
api_resource=True, callcreate_crud_routes() - Register all routes with Litestar
- Return Router instance
5. RPC System
5.1 Whitelisted Controller Methods
-
Create
@whitelistdecorator:- Mark controller methods as publicly callable
- Store metadata on method (
WHITELIST_ATTR) -
is_whitelisted()helper function -
get_whitelist_options()helper function
-
Create
POST /api/v1/rpc/{doctype}/{method}:- Load DocType controller
- Check if method has
@whitelistdecorator - Validate method exists
- Parse request body as method arguments
- Instantiate controller with document (if doc_id provided)
- Call method
- Return result
5.2 Arbitrary RPC Functions
-
Create
@rpcdecorator for standalone functions:- Register function in RPC registry (
RpcRegistrysingleton) - Store function path (e.g.,
my_app.api.send_email) -
is_rpc_function()helper -
get_rpc_options()helper
- Register function in RPC registry (
-
Create
POST /api/v1/rpc/fn/{dotted.path}:- Parse dotted path (e.g.,
my_app.api.send_email) - Look up function in registry
- Validate function has
@rpcdecorator - Parse request body as function arguments
- Call function
- Return result
- Parse dotted path (e.g.,
-
Add permission check for RPC:
- Support
@rpc(permission="custom_permission") - Check permission before calling function
- Support
@rpc(allow_guest=True)for public endpoints
- Support
6. Metadata API
-
Create
GET /api/meta/{doctype}:- Get DocType class from registry
- Generate JSON Schema from Pydantic model:
schema = DocType.model_json_schema() - Extract layout from
Meta.layout - Extract permissions from
Meta.permissions - Return:
{
"doctype": "...",
"schema": {...},
"layout": {...},
"permissions": {...},
"metadata": {...}
}
-
Add field metadata:
- Field labels (from
Field(title=...)) - included in JSON Schema - Help text (from
Field(description=...)) - included in JSON Schema - Validation rules (from Pydantic validators) - included in JSON Schema
- Field types and options - included in JSON Schema
- Field labels (from
7. OpenAPI Documentation
-
Configure Litestar OpenAPI:
- Set title, version, description
- Add authentication scheme (header-based:
x-user-id,x-roles) - Enable Swagger UI at
/schema/swagger - Enable ReDoc at
/schema/redoc
-
Enhance auto-generated docs:
- Add descriptions to CRUD endpoints (via docstrings)
- Add examples for request/response (via Pydantic models)
- Document RPC endpoints (auto-generated from route handlers)
- Add permission requirements to docs (via security schemes)
8. Error Handling
-
Create exception handlers:
-
ValidationError→ 400 Bad Request (validation_error_handler) -
PermissionDeniedError→ 403 Forbidden (permission_denied_handler) -
DocTypeNotFoundError→ 404 Not Found (not_found_handler) -
DuplicateNameError→ 409 Conflict (duplicate_name_handler) - Generic exceptions → 500 Internal Server Error (
framework_error_handler)
-
-
Return consistent error format:
{
"error": "PermissionDenied",
"message": "You don't have permission to delete this document",
"details": {...}
}
9. Request/Response DTOs
-
Use Litestar DTOs for validation:
- Auto-generate from Pydantic models (via Litestar)
- Add DTO for list response (pagination):
PaginatedResponse[T] - Add DTO for error responses:
ErrorResponse
-
Implement response filtering:
- Exclude sensitive fields based on permissions (via field selection)
- Support field selection via query param:
?fields=name,title
10. Testing
10.1 Unit Tests
-
Test permission system:
- Test role-based access (
test_rbac_permission.py) - Test RLS filter generation (
test_rls.py) - Test object-level permissions (
test_object_level_permissions.py)
- Test role-based access (
-
Test RPC system:
- Test
@whitelistdecorator (test_whitelist_decorator.py) - Test
@rpcdecorator (test_rpc_decorator.py) - Test dotted path resolution (
test_rpc_routes.py)
- Test
10.2 Integration Tests
-
Test CRUD endpoints (
test_api_endpoints.py):- Test create with valid data
- Test create with invalid data (validation)
- Test create without permission (403)
- Test list with RLS (only see own docs)
- Test update immutable doc (covered by existing
test_meta_router.py) - Test delete with permission (covered by existing permission tests)
-
Test RPC endpoints (
test_api_endpoints.py,test_rpc_routes.py):- Test whitelisted controller method
- Test arbitrary RPC function
- Test RPC with permissions
-
Test with real HTTP client (
test_real_http_client.py):- Uses
httpx.AsyncClientagainst real uvicorn server - Tests: list endpoint, create endpoint, health check
- Manual testing scripts available in
examples/folderasync with httpx.AsyncClient() as client:
response = await client.post(
"http://localhost:8000/api/v1/todo",
json={"title": "Test"},
headers={"x-user-id": "user123", "x-roles": "Employee"}
)
assert response.status_code == 201
- Uses
Validation Checklist
Before moving to Phase 04, verify:
- All CRUD endpoints work with permissions (verified via
test_meta_router.py,test_api_endpoints.py) - RLS filters are applied correctly (verified via
test_rls.py,test_object_level_permissions.py) - RPC system works for controller methods and functions (verified via
test_rpc_routes.py) - OpenAPI docs are generated correctly (SwaggerUI + ReDoc enabled)
- Error handling returns proper status codes (verified via
test_app.py) - Integration tests pass with real HTTP requests (verified via
test_real_http_client.py)
Anti-Patterns to Avoid
❌ Don't: Hardcode permission checks in endpoints
✅ Do: Use PermissionProtocol for all checks
❌ Don't: Return all data without RLS ✅ Do: Always apply permission filters in repository
❌ Don't: Use global state for current user
✅ Do: Pass UserContext via dependency injection
❌ Don't: Mix authorization logic with business logic ✅ Do: Keep permissions in separate adapter
❌ Don't: Add ignore_permissions=True parameter (Frappe anti-pattern)
✅ Do: Use SystemContext for elevated operations, or mark DocType as requires_auth=False
❌ Don't: Use sudo() or implicit elevation (Odoo anti-pattern)
✅ Do: Explicit system_context() with audit logging
❌ Don't: Return linked documents without permission check (data leakage) ✅ Do: Check permission on Link fields before including related data in response
❌ Don't: Allow bulk delete/update to bypass RLS
✅ Do: Apply RLS filters to all bulk operations (DELETE WHERE ..., UPDATE WHERE ...)
Edge Cases to Handle
Child Table Permissions
- Define: Child tables inherit parent's RLS by default (
Meta.is_child_table = True) - Child rows are not independently permission-checked (
api_resource = False) - Loading parent loads all its children (no separate RLS query)
- Document: If independent child access needed, make it a separate DocType (tests in
test_child_table_permissions.py)
Link Field Data Leakage
- When serializing a doc with Link fields, do NOT auto-include linked doc data (Link fields store UUID, not embedded objects)
- If linked doc data is needed, make a separate API call (permission checked) - documented pattern
- Optional:
?expand=customerparam design documented (checks permission before expanding)
Bulk Operations & RLS
-
GenericRepository.delete_many_for_user(filters)applies RLS -
GenericRepository.update_many_for_user(filters, data)applies RLS - User can only bulk-modify docs they have access to (tests in
test_bulk_operations_rls.py)