Skip to content

Overview

The DatabaseManager (perseus.datamanager.DatabaseManager) handles all requests to the MongoDB instance connected to PERSEUS. Please do not connect directly to the database but only use the DatabaseManager to avoid accidental data corruption.

Every class that the database holds objects from has to inherit from the class perseus.datamanager.DatabaseItem as well as be decorated with @dataclasses.dataclass(kw_only=True). This way, the DatabaseManager class knows how to handle these objects and will store / fetch them correctly. If you want to create a new class that can have objects in the database, you simply have to keep these things in mind when implementing your class. The class attributes can have the following types:

  • str
  • int
  • float
  • bool
  • list[str | int | float | bool | list]
  • dict: Keys need to be str, values can be any of the above

When using only the above types, you just need to define them like this and your class is ready to use:

from dataclasses import dataclass
from perseus.datamanager.DatabaseItem import DatabaseItem
@dataclass(kw_only=True)
class MyDatabaseClass(DatabaseItem):
some_id: str
name: str
value: int
my_list: list[str | float]
my_dict: [str, str]

You can also use other classes as attribute types to allow for more complexity. These attributes require a little help when storing or fetching it from the database. For an example, take a look at the built-in Source class which stores a datetime object:

from dataclasses import dataclass
from datetime import datetime
from typing import Any
from perseus.datamanager import DatabaseItem
@dataclass(kw_only=True)
class Source(DatabaseItem):
"""
This class represents the source of a project fetch.
:param name: The name of the source
:type name: str
:param foreign_id: The id provided from the source
:type foreign_id: int | str
:param created: The creation date and time of the element by the source
:type created: datetime
:param raw: The raw data that the source sent
:type raw: dict[str, Any]
"""
name: str
foreign_id: int | str
created: datetime
is_followup: bool = False
predecessor_id: str | None = None
raw: dict[str, Any]
def db_repr(self) -> dict[str, Any]:
data: dict[str, Any] = self.__dict__.copy()
data["created"] = self.created.isoformat()
return data
@classmethod
def load(cls, **kwargs) -> "Source":
kwargs["created"] = datetime.fromisoformat(kwargs["created"])
return Source(**kwargs)

Here, the methods db_repr() and load() are overwritten to allow for the attribute created to be handled correctly.

If you want to include your new class with API endpoints within a new service, it can be beneficial to overwrite the openapi_documentation_property() method to later use it for an API endpoint documentation. For the Source class from above, this would look like this:

...
from perseus.documentation.EndpointDocumentationProperty import (
EndpointDocumentationProperty,
)
from perseus.documentation.EndpointDocumentationPropertyType import (
EndpointDocumentationPropertyType,
)
...
@classmethod
def openapi_documentation_property(
cls,
include_id: bool = True,
include_files: bool = False,
include_file_tags: bool = False,
accept_null: bool = False,
property_name: str | None = None,
) -> EndpointDocumentationProperty:
base = Source._basic_openapi_documentation_property(
include_id=include_id,
include_files=include_files,
include_file_tags=include_file_tags,
accept_null=accept_null,
property_name=property_name,
)
base.properties.extend(
[
EndpointDocumentationProperty(
"name", EndpointDocumentationPropertyType.STRING
),
EndpointDocumentationProperty(
"foreign_id",
EndpointDocumentationPropertyType.STRING,
EndpointDocumentationPropertyType.NUMBER,
),
EndpointDocumentationProperty(
"created", EndpointDocumentationPropertyType.STRING
),
EndpointDocumentationProperty(
"is_followup", EndpointDocumentationPropertyType.BOOLEAN
),
EndpointDocumentationProperty(
"predecessor_id",
EndpointDocumentationPropertyType.STRING,
EndpointDocumentationPropertyType.NULL,
),
EndpointDocumentationProperty(
"raw", EndpointDocumentationPropertyType.OBJECT
),
]
)
return base

You can use the class DatabaseManager to do the following: