Part 4: Router
We add a minimal router with two endpoints for manual interaction with the pilot system. This lets administrators submit pilots and check status through the HTTP API, complementing the automated task-based submission.
How DiracX discovers routers
DiracX uses entry points to discover routers at startup. The entry point
name determines the URL prefix: a service registered as my_pilots is
mounted at /api/my_pilots/. This convention means the entry point name
is the single source of truth for URL routing.
Implementation
"""Minimal router for manual pilot submission and summary.
Demonstrates:
- Access policy definition
- Dependency injection for MyPilotDB
- Two endpoints for manual interaction
"""
from __future__ import annotations
from collections.abc import Callable
from enum import StrEnum, auto
from typing import Annotated
from diracx.routers.access_policies import BaseAccessPolicy
from diracx.routers.fastapi_classes import DiracxRouter
from diracx.routers.utils.users import AuthorizedUserInfo
from fastapi import Depends
from gubbins.db.sql import MyPilotDB as _MyPilotDB
class ActionType(StrEnum):
CREATE = auto()
READ = auto()
class MyPilotsAccessPolicy(BaseAccessPolicy):
@staticmethod
async def policy(
policy_name: str,
user_info: AuthorizedUserInfo,
/,
*,
action: ActionType | None = None,
):
assert action, "action is a mandatory parameter"
CheckMyPilotsPolicyCallable = Annotated[Callable, Depends(MyPilotsAccessPolicy.check)]
MyPilotDB = Annotated[_MyPilotDB, Depends(_MyPilotDB.transaction)]
router = DiracxRouter()
@router.post("/submit/{ce_name}")
async def submit_pilot(
my_pilot_db: MyPilotDB,
ce_name: str,
check_permission: CheckMyPilotsPolicyCallable,
) -> dict:
await check_permission(action=ActionType.CREATE)
pilot_id = await my_pilot_db.submit_pilot(ce_name)
return {"pilot_id": pilot_id}
@router.get("/summary")
async def get_pilot_summary(
my_pilot_db: MyPilotDB,
check_permission: CheckMyPilotsPolicyCallable,
) -> dict[str, int]:
await check_permission(action=ActionType.READ)
return await my_pilot_db.get_pilot_summary()
Access policy
MyPilotsAccessPolicy controls who can call these endpoints. The
policy method is a @staticmethod with positional-only args
(policy_name, user_info) and keyword-only args with defaults.
Why keyword-only args with defaults?
This follows the Liskov Substitution Principle — subclasses can add new keyword arguments without breaking the base class interface. Every policy parameter must have a default value so that the base class can call the method without knowing about extension-specific parameters. See Security policies for the full reference.
Here we allow all authenticated users — in a real system you'd check
user_info properties to enforce authorization.
DB dependency
The Annotated[DB, Depends(DB.transaction)] pattern
This is the standard way to inject a database connection in DiracX
routers. Depends(DB.transaction) opens a transaction when the request
starts and commits it on success (or rolls back on error). The
Annotated type alias makes the dependency reusable across multiple
endpoint signatures.
Endpoints
- POST
/submit/{ce_name}— Directly inserts a pilot submission into the database (bypassing the task system for manual use). - GET
/summary— Returns pilot counts grouped by status.
For the full guide on building routes, see Add a route and the Routes explanation.
Register entry points
Service entry point:
Access policy entry point:
Update package exports
Re-export the DB dependency:
from gubbins.tasks.depends import MyPilotDB as MyPilotDB # noqa: F401, E402
__all__ += ("MyPilotDB",) # type: ignore[assignment]
Checkpoint
Verify the router works: