Create a project programmatically

Note

This guide assumes you know how to create a new state. If you don’t, please check out how to create a state first.

Although there is the possibility to create a project directly in PERSEUS, new projects will often be created outside of PERSEUS, e.g. in an application form on a website or in other tools.

How to init creation of new project

It is good practice to add a cronjob to the initial state to poll other systems for new project data. When there are new projects in the external system, you can then work with this new data:

from perseus.utils import add_cronjob

from ..BaseState import BaseState

class InitialState(BaseState):
    @classmethod
    def get_task_ids(cls) -> list[str]:
        return ["CHECKING"]

    @classmethod
    def initialize(cls):
        ...

        def fetch_new_projects():
            new_projects = call_external_api_for_new_projects()
            ...

        add_cronjob(fetch_new_projects, "*/15 * * * *") # fetch new projects every 15 minutes

To can read the documentation for more details on how to create and modify a state.

Choose a state machine

Before you can create a new project with the data fetched from an external source, you must think about the state machine that should be used by the new project. You can create a state machine programmatically or use a state machine that was created by the State Machine Manager.

Create state machine programmatically

It is very easy to create a new state machine in your cronjob. Keep in mind to create the states you use in your state machine:

from perseus.utils import add_cronjob
from perseus.state_machine import StateMachine, Path

from ..BaseState import BaseState

class InitialState(BaseState):
    @classmethod
    def get_task_ids(cls) -> list[str]:
        return ["CHECKING"]

    @classmethod
    def initialize(cls):
        ...

        def fetch_new_projects():
            new_projects = call_external_api_for_new_projects()
            ...
            state_machine = StateMachine(Path(("InitialState", None), ("Archive", None)))

In real life, your state machine will often be more complex. It is good practice to create your state machine code in another file and import a method to create the state machine into your InitialState:

create_state_machine.py

from perseus.state_machine import StateMachine, Path, Choice, Fork


def create_state_machine() -> StateMachine:
    """
    Creates a new state machine

    :return: The state machine
    :rtype: StateMachine
    """

    choice_path_1 = Path(("B", None))
    choice_path_2 = Path(("C", ["ALTERNATIVE_PATH", "ANOTHER_ALTERNATIVE_CONDITION"]))
    choice = Choice(choice_path_1, choice_path_2)

    fork_path_1 = Path(("E", None))
    fork_path_2 = Path(("F", None))
    fork = Fork(fork_path_1, fork_path_2)

    top_level_path = Path(("A", None), (choice, None), ("D", None), (fork, None), ("G", None))
    state_machine = StateMachine(top_level_path)

    return state_machine

InitialState.py

from perseus.utils import add_cronjob
from perseus.state_machine import StateMachine, Path
from create_state_machine import create_state_machine

from ..BaseState import BaseState

class InitialState(BaseState):
    @classmethod
    def get_task_ids(cls) -> list[str]:
        return ["CHECKING"]

    @classmethod
    def initialize(cls):
        ...

        def fetch_new_projects():
            new_projects = call_external_api_for_new_projects()
            ...
            state_machine = create_state_machine()

Use state machine from State Machine Manager

Another way of creating a state machine is through the State Machine Manager. To load a state machine created with the manager, you need the ID of the state machine you want to use. You can then load the state machine from the database in your InitialState:

from perseus.utils import add_cronjob

from ..BaseState import BaseState
from perseus.datamanager import DatabaseManager, StoredStateMachine

class InitialState(BaseState):
    @classmethod
    def get_task_ids(cls) -> list[str]:
        return ["CHECKING"]

    @classmethod
    def initialize(cls):
        ...

        def fetch_new_projects():
            new_projects = call_external_api_for_new_projects()
            ...
            state_machine_id = "MY_STATEMACHINE"

            db_manager = DatabaseManager()
            db_result = db_manager.get_item(
                StoredStateMachine, search_filter={"state_machine_id": state_machine_id}
            )
            if db_result is not None:
                state_machine = db_result.state_machine
            else:
                # handle the case that no state machine is stored for your id
                # fall back to a state machine which is created programmatically or the DEFAULT state

Create source

Each project needs a source. The source represents the origin of the project. It is important to set the source so that you can later create reports analyzing where your projects come from.

A source can be easily created on the fly in your state:

from perseus.utils import add_cronjob

from ..BaseState import BaseState

class InitialState(BaseState):
    @classmethod
    def get_task_ids(cls) -> list[str]:
        return ["CHECKING"]

    @classmethod
    def initialize(cls):
        ...

        def fetch_new_projects():
            new_projects = call_external_api_for_new_projects()
            ...
            source = Source(
                name="MY_EXTERNAL_PLATFORM",
                foreign_id="XY-2025-123",
                is_followup=False,
                created=datetime.now(tz=timezone.utc),
                raw={},
                predecessor_id=None,
            )

Add data deletion periods

To comply with the GDPR it is necessary to add data deletion periods to the projects. The Data Deletion Manager can then handle the timely removal of the data.

You can define which data should be deleted, from which state the deletion period should start, and how long the period lasts until the data is deleted.

from perseus.utils import add_cronjob
from perseus.datamanager import DataDeletionPeriod, DataDeletionKey
from ..BaseState import BaseState

class InitialState(BaseState):
    @classmethod
    def get_task_ids(cls) -> list[str]:
        return ["CHECKING"]

    @classmethod
    def initialize(cls):
        ...

        def fetch_new_projects():
            new_projects = call_external_api_for_new_projects()
            ...
            data_deletion_periods = [
                DataDeletionPeriod(
                    state_id="ARCHIVE",
                    key=DataDeletionKey.SOURCE_RAW,
                    additional_period=timedelta(days=93),
                ),
                DataDeletionPeriod(
                    state_id="ARCHIVE",
                    key=DataDeletionKey.PERSON_OF_CONTACT,
                    additional_period=timedelta(days=93),
                ),
            ]

Add requested resources and limits

Each project can request specific resources and limits. The resources and limits the project requests are often part of the proposal. You can add these to the project so that it is available in the process.

from perseus.utils import add_cronjob
from perseus.datamanager import ResourceValue, LimitValue
from ..BaseState import BaseState
from datetime import datetime

class InitialState(BaseState):
    @classmethod
    def get_task_ids(cls) -> list[str]:
        return ["CHECKING"]

    @classmethod
    def initialize(cls):
        ...

        def fetch_new_projects():
            new_projects = call_external_api_for_new_projects()

            requested_resources = [
                ResourceValue(
                    resource_id="hades_cpu_core_h",
                    value=42000000,
                    start=datetime.fromisoformat("2026-01-01T00:00:42+00:00"),
                    end=datetime.fromisoformat("2026-12-31T23:59:42+00:00"),
                ),
                ResourceValue(
                    resource_id="hades_a100_h",
                    value=230000,
                    start=datetime.fromisoformat("2026-01-01T00:00:42+00:00"),
                    end=datetime.fromisoformat("2026-12-31T23:59:42+00:00"),
                )
            ]

            requested_limits = [
                LimitValue(
                    limit_id="ram_per_core",
                    value=192,
                    start=pdatetime.fromisoformat("2026-01-01T00:00:42+00:00"),
                    end=datetime.fromisoformat("2026-12-31T23:59:42+00:00"),
                ),
                LimitValue(
                    limit_id="cores_per_job",
                    value=64,
                    start=datetime.fromisoformat("2026-01-01T00:00:42+00:00"),
                    end=datetime.fromisoformat("2026-12-31T23:59:42+00:00"),
                )
            ]

Timezone in datetime objects

All datetime objects in PERSEUS must be timezone-aware. Usually it is good practice to work with datetime objects in the UTC timezone. See the start and end dates in the example above for the usage of timezone-aware objects.

Create project

When you have your external project data in place alongside your source and state machine, you can then create a new project.

You can read more about the properties of a project in our concept documentation.

from perseus.utils import add_cronjob
from perseus.datamanager import DatabaseManager
from datetime import datetime

from ..BaseState import BaseState

class InitialState(BaseState):
    @classmethod
    def get_task_ids(cls) -> list[str]:
        return ["CHECKING"]

    @classmethod
    def initialize(cls):
        ...

        def fetch_new_projects():
            new_projects = call_external_api_for_new_projects()
            ...
            db_manager = DatabaseManager()

            new_project = Project(
                abbreviation="HPC-PRF-NEW",
                title="My new Project",
                description="The project description",
                project_type="small",
                call="2025/03",
                source=source,
                scientific_fields=[],
                start=datetime.fromisoformat("2026-01-01T00:00:42+00:00"),
                end=datetime.fromisoformat("2026-12-31T23:59:42+00:00"),
                state_machine=state_machine,
                principal_investigator_id=None,
                person_of_contact_id=None,
                affiliation_id=None,
                requested_resources=requested_resources,
                requested_limits=requested_limits,
                data_deletion_periods=data_deletion_periods,
            )

            # save the new project to database
            project_id = db_manager.add_item(new_project)

Full example

Here is a full example of an InitialState where you fetch data from an external source, get a stored state machine, and create a project:

InitialState.py

from perseus.utils import add_cronjob
from perseus.state_machine import StateMachine, Path
from perseus.datamanager import DatabaseManager, Project, Source, StoredStateMachine, DataDeletionPeriod, DataDeletionKey, ResourceValue, LimitValue
from create_state_machine import create_state_machine

from ..BaseState import BaseState

class InitialState(BaseState):
    @classmethod
    def get_task_ids(cls) -> list[str]:
        return ["CHECKING"]

    @classmethod
    def initialize(cls):
        ...

        def fetch_new_projects():
            # in this example call_external_api_for_new_projects()
            # returns an array with objects of a custom ExternalProject class
            new_projects = call_external_api_for_new_projects()

            for project in new_projects:
                # run sanity checks on data

                # gather data from other sources

                state_machine_id = "MY_STATEMACHINE"

                db_manager = DatabaseManager()
                db_result = db_manager.get_item(
                    StoredStateMachine, search_filter={"state_machine_id": state_machine_id}
                )
                if db_result is not None:
                    state_machine = db_result.state_machine
                else:
                    state_machine = StateMachine(Path(("InitialState", None), ("Archive", None)))

                source = Source(
                    name="MY_EXTERNAL_PLATFORM",
                    foreign_id=project.foreign_id,
                    is_followup=False,
                    created=project.create_date,
                    raw={
                        "d_projekt_antrag": project.general_data
                    },
                    predecessor_id=None,
                )

                requested_resources = [
                    ResourceValue(
                        resource_id="hades_cpu_core_h",
                        value=42000000,
                        start=project.resource_start,
                        end=project.resource_end,
                    ),
                    ResourceValue(
                        resource_id="hades_a100_h",
                        value=230000,
                        start=project.resource_start,
                        end=project.resource_end,
                    )
                ]

                requested_limits = [
                    LimitValue(
                        limit_id="ram_per_core",
                        value=192,
                        start=project.limit_start,
                        end=project.limit_end,
                    ),
                    LimitValue(
                        limit_id="cores_per_job",
                        value=64,
                        start=project.limit_start,
                        end=project.limit_end,
                    )
                ]

                data_deletion_periods = [
                    DataDeletionPeriod(
                        state_id="ARCHIVE",
                        key=DataDeletionKey.SOURCE_RAW,
                        additional_period=timedelta(days=93),
                    ),
                    DataDeletionPeriod(
                        state_id="ARCHIVE",
                        key=DataDeletionKey.PERSON_OF_CONTACT,
                        additional_period=timedelta(days=93),
                    ),
                ]

                db_manager = DatabaseManager()

                new_project = Project(
                    abbreviation=project.code,
                    title=project.title,
                    description=project.text,
                    project_type=project.type,
                    call=project.call,
                    source=source,
                    scientific_fields=project.fields,
                    start=project.start_date,
                    end=project.start_date,
                    state_machine=state_machine,
                    custom_fields=project.custom,
                    requested_resources=requested_resources,
                    requested_limits=requested_limits,
                    data_deletion_periods=data_deletion_periods,
                )

                # save the new project to database
                project_id = db_manager.add_item(new_project)


        add_cronjob(fetch_new_projects, "*/15 * * * *") # fetch new project every 15 minutes