PERSEUS core

PERSEUS is a compute project management software for scientific HPC centers. It allows you to

• professionalize your workflows
• deploy center-wide automation
• fully customize workflow items (states), (micro)services and reports

To learn more about PERSEUS, please visit the following:

• PERSEUS main repository
• Documentation
• Live demo

Development

This project uses uv for dependency management.

Install the locked dependencies:

uv sync

Add a runtime dependency:

uv add <package>

Add a development dependency:

uv add --dev <package>

Dependency injection

This project uses Dependency Injector to keep object construction in one place. Instead of creating services directly in route handlers, define them in our container and inject them where they are used.

The important part is that services can have their own constructor dependencies. For example, DatabaseTicketManager needs a DatabaseManager:

class DatabaseTicketManager(TicketManager):
    def __init__(self, db_manager: DatabaseManager):
        self.db_manager = db_manager

Without a container, every caller would need to know how to build the complete object graph:

db_manager = DatabaseManager()
ticket_manager = DatabaseTicketManager(db_manager=db_manager)

The container describes that object graph once:

from dependency_injector import containers, providers


class Container(containers.DeclarativeContainer):
    db_manager = providers.Factory(DatabaseManager)

    ticket_manager = providers.Factory(
        DatabaseTicketManager,
        db_manager=db_manager,
    )

db_manager is a provider for DatabaseManager. ticket_manager is a provider for DatabaseTicketManager, and db_manager=db_manager tells Dependency Injector to call the db_manager provider and pass the result into DatabaseTicketManager.__init__.

So requesting Container.ticket_manager produces the equivalent of:

DatabaseTicketManager(db_manager=DatabaseManager())

Use Provide in FastAPI dependencies and decorate the endpoint with @inject:

from typing import Annotated

from dependency_injector.wiring import Provide, inject
from fastapi import Depends


@inject
def get_ticket(
        ticket_manager: Annotated[
            TicketManager,
            Depends(Provide[Container.ticket_manager]),
        ],
):
    return ticket_manager.find(...)

Outside FastAPI endpoints, use Provide directly without Depends(...):

@inject
def initialize_permissions(
        db_manager: DatabaseManager = Provide[Container.db_manager],
):
    ...

Depends(...) is FastAPI-specific and should only be used at FastAPI boundaries, such as endpoint functions or FastAPI dependency functions. Services and states should usually receive their dependencies through their constructors and be built by the container:

class ProjectState:
    def __init__(self, ticket_manager: TicketManager):
        self.ticket_manager = ticket_manager


class Container(DeclarativeContainer):
    db_manager = Factory(DatabaseManager)

    ticket_manager = Factory(
        DatabaseTicketManager,
        db_manager=db_manager,
    )

    project_state = Factory(
        ProjectState,
        ticket_manager=ticket_manager,
    )

An endpoint can then ask for the top-level object it needs:

@inject
def get_project_state(
        project_state: Annotated[
            ProjectState,
            Depends(Provide[Container.project_state]),
        ],
):
    return project_state.render(...)

It is also technically possible to use @inject and Provide[...] in any wired function, including functions in service or state modules. Prefer the constructor-provider style above for services and states because it keeps object construction centralized and keeps the classes independent of FastAPI and the DI framework.

The container must then be wired to the module that contains the injected function. Wiring is what replaces Provide[Container.ticket_manager] with the real DatabaseTicketManager instance:

container = Container()
container.wire(packages=["perseus.api"])

For tests, override providers on the app container:

ticket_manager_mock = mock.Mock(spec=TicketManager)

with app.container.ticket_manager.override(ticket_manager_mock):
    response = client.get("/service/TicketManager/ticket/...")

Inside that with block, endpoints that ask for Container.ticket_manager receive ticket_manager_mock instead of a DatabaseTicketManager. Have a look at tests/api/test_tickets.py for a complete example.

License

This project is licensed under the terms of the MIT License. See the LICENSE file for details.