Server Events¶

AMSDAL Server emits events at key points during the server lifecycle and request processing. Plugins and application code can listen to these events to add custom behavior.

All server events use the Events System. Each event has a typed context containing the data relevant to that point in the lifecycle.

Lifecycle Events¶

ServerStartupEvent¶

Emitted when the server starts up (in the FastAPI lifespan context). Use this for initialization tasks.

from amsdal_utils.events import EventListener, listen_to
from amsdal_server.apps.common.events.server import ServerStartupEvent, ServerStartupContext

@listen_to(ServerStartupEvent)
class MyStartupListener(EventListener[ServerStartupContext]):
    def handle(self, context: ServerStartupContext, next_fn):
        app = context.app  # FastAPI instance
        # Initialization logic here
        return next_fn(context)

    async def ahandle(self, context: ServerStartupContext, next_fn):
        app = context.app
        # Async initialization logic
        return await next_fn(context)

Context fields: app: FastAPI

ServerShutdownEvent¶

Emitted when the server shuts down. Use this for cleanup tasks.

from amsdal_server.apps.common.events.server import ServerShutdownEvent, ServerShutdownContext

@listen_to(ServerShutdownEvent)
class MyShutdownListener(EventListener[ServerShutdownContext]):
    async def ahandle(self, context: ServerShutdownContext, next_fn):
        # Close connections, flush caches, etc.
        return await next_fn(context)

Context fields: app: FastAPI

Setup Events¶

RouterSetupEvent¶

Emitted after standard routes are registered. Use this to add custom API endpoints.

from fastapi import APIRouter
from amsdal_server.apps.common.events.server import RouterSetupEvent, RouterSetupContext

@listen_to(RouterSetupEvent)
class MyRoutesListener(EventListener[RouterSetupContext]):
    def handle(self, context: RouterSetupContext, next_fn):
        router = APIRouter()

        @router.get('/api/custom/hello')
        async def hello():
            return {'message': 'Hello from plugin!'}

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

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

Context fields: app: FastAPI

Note

RouterSetupEvent is emitted synchronously via EventBus.emit(). Implement the handle() method.

MiddlewareSetupEvent¶

Emitted after standard middlewares are registered. Use this to add custom middleware.

from amsdal_server.apps.common.events.server import MiddlewareSetupEvent, MiddlewareSetupContext

@listen_to(MiddlewareSetupEvent)
class MyMiddlewareListener(EventListener[MiddlewareSetupContext]):
    def handle(self, context: MiddlewareSetupContext, next_fn):
        context.app.add_middleware(MyCORSMiddleware, allow_origins=['*'])
        return next_fn(context)

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

Context fields: app: FastAPI

Note

MiddlewareSetupEvent is emitted synchronously via EventBus.emit().

Server Initialization Flow¶

The events are emitted in this order during build_app():

Plugin loading — AppConfig.on_setup() for each contrib in AMSDAL_CONTRIBS
Standard routes registered
RouterSetupEvent — plugins add custom routes
Standard middleware registered
MiddlewareSetupEvent — plugins add custom middleware
Server starts → ServerStartupEvent (async, in lifespan)
Server running...
Server stops → ServerShutdownEvent (async, in lifespan)

Pre-Response Events¶

Pre-response events are emitted just before the server returns data to the client. Use them to modify or enrich the response.

Object Events¶

from amsdal_server.apps.objects.events.pre_response import (
    ObjectListPreResponseEvent,
    ObjectListPreResponseContext,
    ObjectDetailPreResponseEvent,
    ObjectDetailPreResponseContext,
)

ObjectListPreResponseEvent — emitted before returning an object list.

Field	Type	Description
`request`	`Request`	The incoming HTTP request
`class_name`	`str`	Name of the model class
`response`	`_ObjectListResponse`	The response data (modifiable)

ObjectDetailPreResponseEvent — emitted before returning an object detail.

Field	Type	Description
`request`	`Request`	The incoming HTTP request
`class_name`	`str`	Name of the model class
`address`	`str`	Object address
`response`	`_ObjectDetailResponse`	The response data (modifiable)

Class Events¶

from amsdal_server.apps.classes.events.pre_response import (
    ClassListPreResponseEvent,
    ClassListPreResponseContext,
    ClassDetailPreResponseEvent,
    ClassDetailPreResponseContext,
)

ClassListPreResponseEvent — emitted before returning the class list.

ClassDetailPreResponseEvent — emitted before returning class detail.

Transaction Events¶

from amsdal_server.apps.transactions.events.pre_response import (
    TransactionListPreResponseEvent,
    TransactionListPreResponseContext,
    TransactionDetailPreResponseEvent,
    TransactionDetailPreResponseContext,
)

TransactionListPreResponseEvent — emitted before returning the transaction list.

TransactionDetailPreResponseEvent — emitted before returning transaction detail.

Example: Enriching Object List Response¶

@listen_to(ObjectListPreResponseEvent)
class AddCustomColumnsListener(EventListener[ObjectListPreResponseContext]):
    async def ahandle(self, context: ObjectListPreResponseContext, next_fn):
        # Add a custom column to the response
        context = context.create_next(
            listener_id=self.listener_id,
            response=context.response.model_copy(
                update={'columns': context.response.columns + [custom_column]}
            ),
        )
        return await next_fn(context)

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

Authentication & Authorization Events¶

AuthenticateEvent¶

Emitted during request authentication. Listeners decode credentials and set the user.

from amsdal_server.apps.common.events.auth import AuthenticateEvent, AuthenticateContext

Field	Type	Description
`auth_header`	`str`	The `Authorization` header value
`connection`	`HTTPConnection`	The HTTP connection
`user`	`BaseUser \\| None`	The authenticated user (set by listener)
`scopes`	`list[str]`	Permission scopes (set by listener)

ClassAuthorizeEvent / ObjectAuthorizeEvent¶

Emitted to check permissions at class or object level.

from amsdal_server.apps.common.events.authorize import (
    ClassAuthorizeEvent,
    ClassAuthorizeContext,
    ObjectAuthorizeEvent,
    ObjectAuthorizeContext,
)

See Auth & Permissions for details.