Create a Plugin

This guide walks you through building an AMSDAL plugin — from defining configuration to adding routes, middleware, models, and event listeners.

Plugin Structure

A typical plugin has this layout:

my_plugin/
├── __init__.py
├── app.py                # AppConfig class
├── models/               # Plugin models
│   ├── __init__.py
│   └── my_model.py
├── transactions/         # Plugin transactions
│   ├── __init__.py
│   └── my_transactions.py
├── event_handlers/       # Event listeners
│   ├── __init__.py
│   └── listeners.py
└── fixtures/             # Default data (optional)
    └── initial.json

Step 1: Define AppConfig

Every plugin needs an AppConfig subclass. The on_setup() method is called when the plugin is loaded.

# my_plugin/app.py
from amsdal.contrib.app_config import AppConfig


class MyPluginAppConfig(AppConfig):
    def on_setup(self) -> None:
        # Register event listeners, configure the plugin
        from amsdal_utils.events import EventBus
        from amsdal_server.apps.common.events.server import (
            RouterSetupEvent,
            ServerStartupEvent,
        )
        from my_plugin.event_handlers.listeners import (
            MyRouteListener,
            MyStartupListener,
        )

        EventBus.subscribe(RouterSetupEvent, MyRouteListener)
        EventBus.subscribe(ServerStartupEvent, MyStartupListener)

Step 2: Register the Plugin

Add your AppConfig class path to the AMSDAL_CONTRIBS setting:

# Environment variable (comma-separated)
AMSDAL_CONTRIBS="amsdal.contrib.auth.app.AuthAppConfig,amsdal.contrib.frontend_configs.app.FrontendConfigAppConfig,my_plugin.app.MyPluginAppConfig"

AMSDAL will import and call on_setup() for each config during application initialization.

Step 3: Add Event Listeners

Use the Events System to hook into the AMSDAL lifecycle.

# my_plugin/event_handlers/listeners.py
from amsdal_utils.events import EventListener, NextFn, AsyncNextFn
from amsdal_server.apps.common.events.server import (
    ServerStartupContext,
)


class MyStartupListener(EventListener[ServerStartupContext]):
    def handle(self, context: ServerStartupContext, next_fn: NextFn):
        print('Plugin initialized!')
        return next_fn(context)

    async def ahandle(self, context: ServerStartupContext, next_fn: AsyncNextFn):
        print('Plugin initialized (async)!')
        return await next_fn(context)

Step 4: Add Custom Routes

Listen to RouterSetupEvent to add API endpoints via FastAPI routers:

from fastapi import APIRouter
from amsdal_utils.events import EventListener, NextFn
from amsdal_server.apps.common.events.server import (
    RouterSetupContext,
)


class MyRouteListener(EventListener[RouterSetupContext]):
    def handle(self, context: RouterSetupContext, next_fn: NextFn):
        router = APIRouter(tags=['my-plugin'])

        @router.get('/api/my-plugin/status')
        async def plugin_status():
            return {'status': 'active'}

        @router.post('/api/my-plugin/action')
        async def plugin_action(data: dict):
            # Handle the action
            return {'result': 'ok'}

        context.app.include_router(router)
        return next_fn(context)

    async def ahandle(self, context, next_fn):
        return self.handle(context, next_fn)

Note

RouterSetupEvent is emitted synchronously. Implement the handle() method.

Step 5: Add Custom Middleware

Listen to MiddlewareSetupEvent to add middleware:

from starlette.middleware.base import BaseHTTPMiddleware
from amsdal_utils.events import EventListener, NextFn
from amsdal_server.apps.common.events.server import (
    MiddlewareSetupContext,
)


class RequestTimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        import time
        start = time.time()
        response = await call_next(request)
        response.headers['X-Process-Time'] = str(time.time() - start)
        return response


class MyMiddlewareListener(EventListener[MiddlewareSetupContext]):
    def handle(self, context: MiddlewareSetupContext, next_fn: NextFn):
        context.app.add_middleware(RequestTimingMiddleware)
        return next_fn(context)

    async def ahandle(self, context, next_fn):
        return self.handle(context, next_fn)

Step 6: Add Models

Place model files in a models/ directory within your plugin package. AMSDAL automatically discovers them.

# my_plugin/models/audit_log.py
from amsdal_models.classes.model import Model
from amsdal_utils.models.enums import ModuleType


class AuditLog(Model):
    __module_type__ = ModuleType.CONTRIB

    action: str
    user_email: str
    details: str
    timestamp: str

Mark contrib models with __module_type__ = ModuleType.CONTRIB so AMSDAL recognizes them as plugin-provided models and handles their migrations.

Step 7: Add Transactions

Place transaction files in a transactions/ directory within your plugin package. AMSDAL automatically discovers them via AST parsing and exposes them through the REST API.

# my_plugin/transactions/audit.py
from amsdal_data.transactions.decorators import transaction
from amsdal.contrib.auth.decorators import require_auth


@require_auth
@transaction(tags=['Audit'])
def export_audit_log(date_from: str, date_to: str) -> dict:
    """Export audit log entries for a date range."""
    from my_plugin.models.audit_log import AuditLog

    entries = list(
        AuditLog.objects.filter(
            timestamp__gte=date_from,
            timestamp__lte=date_to,
        )
    )
    return {'count': len(entries), 'entries': [e.to_dict() for e in entries]}

Once discovered, the transaction is available via REST API:

• GET /api/transactions/ — lists all transactions including plugin ones
• GET /api/transactions/export_audit_log/ — schema and parameter details
• POST /api/transactions/export_audit_log/ — execute with {"args": {"date_from": "2025-01-01", "date_to": "2025-12-31"}}

Use @async_transaction() for async transactions. Auth decorators (@require_auth, @allow_any, @permissions()) must be placed above the transaction decorator.

Note

No explicit registration is needed — AMSDAL scans the transactions/ directory of every registered contrib automatically.

Step 8: Modify Responses (Optional)

Use pre-response events to enrich API responses:

from amsdal_utils.events import EventBus
from amsdal_server.apps.objects.events.pre_response import (
    ObjectListPreResponseEvent,
    ObjectListPreResponseContext,
)


class EnrichListResponseListener(EventListener[ObjectListPreResponseContext]):
    async def ahandle(self, context: ObjectListPreResponseContext, next_fn):
        # Add custom data to the response before it's sent
        return await next_fn(context)

    def handle(self, context, next_fn):
        raise NotImplementedError

Register in on_setup():

EventBus.subscribe(ObjectListPreResponseEvent, EnrichListResponseListener)

---

Complete Example

Here's a minimal but complete plugin:

# my_analytics/app.py
from amsdal.contrib.app_config import AppConfig


class AnalyticsAppConfig(AppConfig):
    def on_setup(self) -> None:
        from amsdal_utils.events import EventBus
        from amsdal_server.apps.common.events.server import RouterSetupEvent

        from my_analytics.routes import AnalyticsRouteListener

        EventBus.subscribe(RouterSetupEvent, AnalyticsRouteListener)

# my_analytics/routes.py
from fastapi import APIRouter
from amsdal_utils.events import EventListener, NextFn
from amsdal_server.apps.common.events.server import RouterSetupContext


class AnalyticsRouteListener(EventListener[RouterSetupContext]):
    def handle(self, context: RouterSetupContext, next_fn: NextFn):
        router = APIRouter(tags=['analytics'])

        @router.get('/api/analytics/summary')
        async def summary():
            return {'total_users': 42, 'active_today': 10}

        context.app.include_router(router)
        return next_fn(context)

    async def ahandle(self, context, next_fn):
        return self.handle(context, next_fn)

Register:

AMSDAL_CONTRIBS="amsdal.contrib.auth.app.AuthAppConfig,amsdal.contrib.frontend_configs.app.FrontendConfigAppConfig,my_analytics.app.AnalyticsAppConfig"

---

Packaging & Distribution

To distribute your plugin as a Python package:

1. Create a standard pyproject.toml with your plugin package
2. Add amsdal as a dependency
3. Document the AMSDAL_CONTRIBS path users need to add
4. Publish to PyPI

[project]
name = "amsdal-my-plugin"
dependencies = ["amsdal>=1.0"]

Users install and register:

pip install amsdal-my-plugin

AMSDAL_CONTRIBS="...,my_plugin.app.MyPluginAppConfig"