Protocol Interfaces
Auto-generated from source code - Do not edit manually.
Framework M uses Protocol interfaces to define contracts between components. Swap implementations by providing different adapters.
AuditLogProtocol
File: core/interfaces/audit.py
Protocol defining the contract for audit log implementations.
Methods:
async def log(user_id: str, action: str, doctype: str, document_id: str, changes: dict[str, Any] | None, metadata: dict[str, Any] | None) -> str- Log an audit entry.
async def query(filters: dict[str, Any] | None, limit: int, offset: int) -> list[AuditEntry]- Query audit entries with filters.
AuthContextProtocol
File: core/interfaces/auth_context.py
Protocol defining the contract for authentication context.
Methods:
async def get_current_user() -> UserContext- Get the current authenticated user from request context.
async def get_user_by_id(user_id: str) -> UserContext | None- Look up a user by their ID.
AuthenticationProtocol
File: core/interfaces/authentication.py
Protocol for pluggable authentication strategies.
Methods:
def supports(headers: Mapping[str, str]) -> bool- Check if this strategy can handle the request.
async def authenticate(headers: Mapping[str, str]) -> UserContext | None- Authenticate the request and extract user identity.
BaseDocTypeProtocol
File: core/interfaces/base_doctype.py
Protocol defining the minimal interface for all DocTypes.
BootstrapProtocol
File: core/interfaces/bootstrap.py
Protocol defining the interface for bootstrap steps.
Methods:
async def run(container: Any) -> None
CacheProtocol
File: core/interfaces/cache.py
Protocol defining the contract for cache implementations.
Methods:
async def get(key: str) -> Any | None- Get a value from cache.
async def set(key: str, value: Any, ttl: int | None) -> None- Set a value in cache.
async def delete(key: str) -> None- Delete a key from cache.
async def exists(key: str) -> bool- Check if a key exists in cache.
async def get_many(keys: list[str]) -> dict[str, Any]- Get multiple values from cache.
async def set_many(items: dict[str, Any], ttl: int | None) -> None- Set multiple values in cache.
async def delete_pattern(pattern: str) -> int- Delete all keys matching a pattern.
async def ttl(key: str) -> int | None- Get remaining TTL for a key.
async def clear() -> None- Clear all keys from cache.
async def acquire_lock(key: str, ttl: int) -> bool- Acquire a distributed lock.
async def release_lock(key: str) -> bool- Release a distributed lock.
BaseControllerProtocol
File: core/interfaces/controller.py
Protocol defining the interface for document lifecycle hooks.
Methods:
async def validate(context: Any | None) -> Noneasync def before_insert(context: Any | None) -> Noneasync def after_insert(context: Any | None) -> Noneasync def before_save(context: Any | None) -> Noneasync def after_save(context: Any | None) -> Noneasync def before_delete(context: Any | None) -> Noneasync def after_delete(context: Any | None) -> Noneasync def on_submit(context: Any | None) -> Noneasync def on_cancel(context: Any | None) -> None
EmailQueueProtocol
File: core/interfaces/email_queue.py
Protocol for email queue operations.
Methods:
async def queue_email(request: EmailRequest) -> EmailQueueResult- Queue an email for sending.
async def get_status(queue_id: str) -> str | None- Get the status of a queued email.
async def cancel(queue_id: str) -> bool- Cancel a queued email (if not yet sent).
EmailSenderProtocol
File: core/interfaces/email_sender.py
Protocol for email sending operations.
Methods:
async def send(message: EmailMessage) -> SendResult- Send an email message.
async def is_available() -> bool- Check if the sender is available.
EventBusProtocol
File: core/interfaces/event_bus.py
Protocol defining the contract for event bus implementations.
Methods:
async def connect() -> None- Establish connection to the message broker.
async def disconnect() -> None- Close connection to the message broker.
def is_connected() -> bool- Check if connected to the message broker.
async def publish(topic: str, event: Event) -> None- Publish an event to a topic.
async def subscribe(topic: str, handler: EventHandler) -> str- Subscribe to events on a specific topic.
async def subscribe_pattern(pattern: str, handler: EventHandler) -> str- Subscribe to events matching a pattern.
async def unsubscribe(subscription_id: str) -> None- Remove a subscription.
I18nProtocol
File: core/interfaces/i18n.py
Protocol defining the contract for internationalization.
Methods:
async def translate(text: str, locale: str | None) -> str- Translate text to the specified locale.
async def get_locale() -> str- Get the current locale for the request context.
async def set_locale(locale: str) -> None- Set the locale for the current request context.
async def get_available_locales() -> list[str]- Get list of available locales.
IdentityProtocol
File: core/interfaces/identity.py
Protocol defining the contract for identity management.
Methods:
async def get_user(user_id: str) -> UserContext | None- Get user by their unique identifier.
async def get_user_by_email(email: str) -> UserContext | None- Get user by their email address.
async def authenticate(credentials: Credentials) -> Token- Authenticate user with provided credentials.
async def get_attributes(user_id: str) -> dict[str, Any]- Get user's ABAC (Attribute-Based Access Control) attributes.
async def validate_token(token: str) -> UserContext | None- Validate an access token and return the associated user.
JobLoggerProtocol
File: core/interfaces/job_logger.py
Protocol for job execution logging.
Methods:
async def log_enqueue(job_id: str, job_name: str) -> JobLog- Log that a job was enqueued.
async def log_start(job_id: str) -> JobLog | None- Log that a job started executing.
async def log_success(job_id: str, result: dict[str, Any] | None) -> JobLog | None- Log that a job completed successfully.
async def log_failure(job_id: str, error: str) -> JobLog | None- Log that a job failed.
async def get_log(job_id: str) -> JobLog | None- Get a job log by ID.
JobQueueProtocol
File: core/interfaces/job_queue.py
Protocol defining the contract for job queue implementations.
Methods:
async def enqueue(job_name: str) -> str- Add a job to the queue for immediate processing.
async def schedule(job_name: str, cron: str) -> str- Schedule a recurring job using cron syntax.
async def cancel(job_id: str) -> bool- Cancel a pending or running job.
async def get_status(job_id: str) -> JobInfo | None- Get current status and info for a job.
async def retry(job_id: str) -> str- Retry a failed job.
NotificationProtocol
File: core/interfaces/notification.py
Protocol defining the contract for notification services.
Methods:
async def send_email(to: list[str], subject: str, body: str) -> bool- Send an email notification.
async def send_sms(to: str, body: str) -> bool- Send an SMS notification.
async def send_push(user_id: str, title: str, body: str) -> bool- Send a push notification.
async def send_in_app(user_id: str, title: str, body: str) -> bool- Send an in-app notification.
OAuth2Protocol
File: core/interfaces/oauth.py
Protocol for OAuth2/OIDC provider implementations.
Methods:
def provider_name() -> str- Return the provider name (e.g., 'google', 'github').
async def get_authorization_url(state: str, redirect_uri: str) -> str- Generate authorization URL to redirect user to provider.
async def exchange_code(code: str, redirect_uri: str) -> OAuth2Token- Exchange authorization code for access token.
async def get_user_info(token: OAuth2Token) -> OAuth2UserInfo- Fetch user information from provider.
PermissionProtocol
File: core/interfaces/permission.py
Protocol defining the contract for permission checking.
Methods:
async def evaluate(request: PolicyEvaluateRequest) -> PolicyEvaluateResult- Evaluate a policy request.
async def get_permitted_filters(principal: str, principal_attributes: dict[str, Any], resource: str, tenant_id: str | None) -> dict[str, Any]- Get filters for Row-Level Security.
Printable
File: core/interfaces/print.py
Protocol for printable documents.
Methods:
def model_dump() -> dict[str, Any]- Return document as dictionary.
PrintProtocol
File: core/interfaces/print.py
Protocol defining the contract for document printing/rendering.
Methods:
async def render(doc: BaseModel, template: str, format: PrintFormat | str, context: dict[str, Any] | None) -> bytes- Render a document to the specified format.
async def get_templates(doctype: str) -> list[str]- Get available print templates for a DocType.
ReadModelProtocol
File: core/interfaces/read_model.py
Protocol defining the contract for CQRS read model implementations.
Methods:
async def project(event: Event) -> None- Process a domain event and update the read model.
async def query(filters: dict[str, Any] | None, order_by: list[str] | None, limit: int, offset: int) -> list[dict[str, Any]]- Query the read model with optional filtering and pagination.
async def rebuild() -> None- Rebuild the entire read model from event history.
ReportEngineProtocol
File: core/interfaces/report_engine.py
Protocol defining the contract for report engine implementations.
Methods:
async def execute(query: str, parameters: dict[str, Any] | None) -> ReportResult- Execute a report query and return results.
async def validate(query: str) -> bool- Validate query syntax without executing.
RepositoryProtocol
File: core/interfaces/repository.py
Protocol defining the contract for data repositories.
Methods:
async def get(id: UUID) -> T | None- Retrieve an entity by its unique identifier.
async def save(entity: T, version: int | None) -> T- Persist an entity (insert or update).
async def delete(id: UUID) -> None- Delete an entity by its unique identifier.
async def exists(id: UUID) -> bool- Check if an entity exists by its identifier.
async def count(filters: list[FilterSpec] | None) -> int- Count entities matching the given filters.
async def list(filters: list[FilterSpec] | None, order_by: list[OrderSpec] | None, limit: int, offset: int) -> PaginatedResult[T]- List entities with filtering, sorting, and pagination.
async def bulk_save(entities: Sequence[T]) -> Sequence[T]- Save multiple entities in a single operation.
SchemaMapperProtocol
File: core/interfaces/schema_mapper.py
Protocol defining the interface for schema mapping.
Methods:
def create_table(model: type[BaseDocTypeProtocol], metadata: Any) -> Anydef create_tables(model: type[BaseDocTypeProtocol], metadata: Any) -> list[Any]def detect_schema_changes(old_model: type[BaseDocTypeProtocol], new_model: type[BaseDocTypeProtocol]) -> dict[str, Any]
SearchProtocol
File: core/interfaces/search.py
Protocol defining the contract for full-text search.
Methods:
async def index(doctype: str, doc_id: str, doc: dict[str, Any]) -> None- Index a document for searching.
async def delete_index(doctype: str, doc_id: str) -> None- Remove a document from the search index.
async def search(doctype: str, query: str) -> SearchResult- Search for documents.
async def reindex(doctype: str) -> int- Rebuild the entire index for a DocType.
SecretProtocol
File: core/interfaces/secrets.py
Protocol for secrets retrieval (Port).
Methods:
def get_secret(key: str, default: str | None) -> str | None- Retrieve a secret by its key.
def list_required_secrets() -> list[str]- Return a list of secret keys managed by this provider.
SessionProtocol
File: core/interfaces/session.py
Protocol for session storage backends.
Methods:
async def create(user_id: str, ip_address: str | None, user_agent: str | None) -> SessionData- Create a new session.
async def get(session_id: str) -> SessionData | None- Get session by ID.
async def delete(session_id: str) -> bool- Delete a session.
async def delete_all_for_user(user_id: str) -> int- Delete all sessions for a user.
async def list_for_user(user_id: str) -> list[SessionData]- List all active sessions for a user.
SocketProtocol
File: core/interfaces/socket.py
Protocol for real-time WebSocket communication.
Methods:
async def broadcast(topic: str, message: dict[str, Any]) -> None- Broadcast a message to all subscribers of a topic.
async def send_to_user(user_id: str, message: dict[str, Any]) -> None- Send a message to a specific user.
StorageProtocol
File: core/interfaces/storage.py
Protocol defining the contract for file storage.
Methods:
async def save_file(path: str, content: bytes, content_type: str | None) -> str- Save file content to storage.
async def get_file(path: str) -> bytes- Retrieve file content from storage.
async def delete_file(path: str) -> None- Delete a file from storage.
async def exists(path: str) -> bool- Check if a file exists in storage.
async def list_files(prefix: str) -> list[str]- List files matching a prefix.
async def get_metadata(path: str) -> FileMetadata | None- Get metadata for a file without fetching content.
async def get_url(path: str, expires: int) -> str- Generate a presigned URL for file access.
async def copy(src: str, dest: str) -> str- Copy a file within storage.
async def move(src: str, dest: str) -> str- Move a file within storage.
TenantProtocol
File: core/interfaces/tenant.py
Protocol for tenant resolution and attribute retrieval.
Methods:
def get_current_tenant() -> str- Get the current tenant ID from request context.
def get_tenant_attributes(tenant_id: str) -> dict[str, Any]- Get attributes for a specific tenant.
WebhookLoaderProtocol
File: core/interfaces/webhook.py
Protocol for loading webhooks from storage.
Methods:
async def load_active_webhooks() -> list[Webhook]- Load all active webhooks.
WebhookDelivererProtocol
File: core/interfaces/webhook.py
Protocol for delivering webhooks.
Methods:
async def deliver(webhook: Webhook, event: Event) -> None- Deliver an event to a webhook endpoint.
WorkflowProtocol
File: core/interfaces/workflow.py
Protocol for workflow state management.
Methods:
async def start_workflow(doctype: str, doc_id: str, workflow_name: str, user: UserContext) -> WorkflowStateInfo- Start a workflow on a document.
async def get_workflow_state(doctype: str, doc_id: str) -> WorkflowStateInfo | None- Get current workflow state for a document.
async def transition(request: TransitionRequest) -> TransitionResult- Attempt to transition a document to a new workflow state.
async def get_available_actions(doctype: str, doc_id: str, user: UserContext) -> list[str]- Get list of actions available to user in current state.