Skip to main content

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-injector Container (wire modules)
    • Middleware Discovery:
      • Scan MiddlewareRegistry for 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).
  • 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-id header
    • Extract x-roles header (comma-separated)
    • Extract x-tenants header (optional, comma-separated)
    • Create UserContext object
    • Store in request.state.user
    • Return 401 if headers missing (configurable)
  • 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
    • CustomPermission DocType created (integration pending)
    • Return PolicyEvaluateResult(authorized=True/False, decision_source=DecisionSource.RBAC)

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_authapply_rlsUse Case
TrueTrueDefault. Alice only sees her own invoices.
TrueFalseLookup tables. All users see all countries.
FalseFalsePublic data. App config, announcements.
FalseTruePublic 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

3.3 System Context (Elevated Operations)

[!IMPORTANT] No ignore_permissions flag. For system operations, use explicit SystemContext.

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
  • 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 PermissionDeniedError if denied

3.5 Team-Based Access (Indie)

For team/department-level access without full ReBAC:

  • Add Meta.rls_field option:
    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

[!NOTE] Where does user_teams come 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 + UserAttribute DocTypes for self-contained apps

3.6 Explicit Sharing (DocumentShare)

For sharing specific documents with specific users/roles:

  • Create built-in DocumentShare DocType:
    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 via get_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 PolicyEvaluateRequest from self.user, self.doctype_name
    • Calls permission.evaluate(request)
    • Raises PermissionDeniedError if not authorized
    • Works on Controller methods

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} → checks CREATE permission (verified via test)
    • GET /api/v1/{doctype}/{id} → checks READ permission (verified via test)
    • PUT /api/v1/{doctype}/{id} → checks WRITE permission (verified via test)
    • DELETE /api/v1/{doctype}/{id} → checks DELETE permission (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()

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 via apply_rls_filters()
    • Return paginated response with metadata:
      {
      "data": [...],
      "total": 100,
      "limit": 20,
      "offset": 0
      }

4.3 Create Endpoint

  • Create POST /api/v1/{doctype}:
    • Validate request body against Pydantic model
    • Check CREATE permission via _check_permission()
    • Set owner to 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 READ permission
    • Return 404 if not found (via NotFoundException)
    • Return 403 if permission denied
    • Return document

4.5 Update Endpoint

  • Create PUT /api/v1/{doctype}/{id}:
    • Load existing document via repo.get()
    • Check WRITE permission
    • Check if submitted (deny if immutable) via _check_submitted()
    • Merge changes via entity.model_copy(update=data)
    • Call repository save()
    • Return updated document

4.6 Delete Endpoint

  • Create DELETE /api/v1/{doctype}/{id}:
    • Load document
    • Check DELETE permission
    • 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, call create_crud_routes()
    • Register all routes with Litestar
    • Return Router instance

5. RPC System

5.1 Whitelisted Controller Methods

  • Create @whitelist decorator:

    • 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 @whitelist decorator
    • 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 @rpc decorator for standalone functions:

    • Register function in RPC registry (RpcRegistry singleton)
    • Store function path (e.g., my_app.api.send_email)
    • is_rpc_function() helper
    • get_rpc_options() helper
  • 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 @rpc decorator
    • Parse request body as function arguments
    • Call function
    • Return result
  • Add permission check for RPC:

    • Support @rpc(permission="custom_permission")
    • Check permission before calling function
    • Support @rpc(allow_guest=True) for public endpoints

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

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 RPC system:

    • Test @whitelist decorator (test_whitelist_decorator.py)
    • Test @rpc decorator (test_rpc_decorator.py)
    • Test dotted path resolution (test_rpc_routes.py)

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.AsyncClient against real uvicorn server
    • Tests: list endpoint, create endpoint, health check
    • Manual testing scripts available in examples/ folder
      async 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

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)
  • 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=customer param 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)