Skip to content

DESDEO REST API

The DESDEO REST API is a FastAPI application that provides a RESTful interface to the DESDEO framework. The API is designed to be used by the DESDEO WebUI, but it can also be used by other applications. The best way to get the API docs is to use the Swagger/Redoc UI provided by the API. You can access the Swagger UI at http://localhost:8000/docs and the Redoc UI at http://localhost:8000/redoc after starting the API. This assumes that the API is running on your local machine and the default port is used.

The Main Application

desdeo.api.app

The main FastAPI application for the DESDEO API.

root async

root() -> dict

Just a simple hello world message.

Source code in desdeo/api/app.py
@app.get("/")
async def root() -> dict:
    """Just a simple hello world message."""
    return {"message": "Hello World!"}

Database initializer

desdeo.api.db_init

This module initializes the database.

Main Database method

desdeo.api.db

Database configuration file for the API.

get_db

get_db() -> Generator[Session, None, None]

Get a database session as a dependency.

Source code in desdeo/api/db.py
def get_db() -> Generator[Session, None, None]:
    """Get a database session as a dependency."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Database Models

desdeo.api.db_models

All models for the API. I put them all in a single file for simplicity.

Log

Bases: Base

A model to store logs of user actions. I have no idea what to put in this table.

Source code in desdeo/api/db_models.py
class Log(Base):
    """A model to store logs of user actions. I have no idea what to put in this table."""

    __tablename__ = "log"
    id: Mapped[int] = mapped_column(primary_key=True, unique=True)
    user = mapped_column(Integer, ForeignKey("user.id"), nullable=False)
    action: Mapped[str] = mapped_column(nullable=False)
    value = mapped_column(JSON, nullable=False)
    timestamp: Mapped[str] = mapped_column(nullable=False)

Method

Bases: Base

A model to store a method and its associated data.

Source code in desdeo/api/db_models.py
class Method(Base):
    """A model to store a method and its associated data."""

    __tablename__ = "method"
    id: Mapped[int] = mapped_column(primary_key=True, unique=True)
    kind: Mapped[schema.Methods] = mapped_column(Enum(schema.Methods), nullable=False)
    properties: Mapped[list[schema.MethodProperties]] = mapped_column(
        ARRAY(Enum(schema.MethodProperties)), nullable=False
    )
    name: Mapped[str] = mapped_column(nullable=False)
    parameters = mapped_column(JSON, nullable=True)

MethodState

Bases: Base

A model to store the state of a method. Contains all the information needed to restore the state of a method.

Source code in desdeo/api/db_models.py
class MethodState(Base):
    """A model to store the state of a method. Contains all the information needed to restore the state of a method."""

    __tablename__ = "method_state"
    id: Mapped[int] = mapped_column(primary_key=True, unique=True)
    user = mapped_column(Integer, ForeignKey("user.id"), nullable=False)
    problem = mapped_column(Integer, ForeignKey("problem.id"), nullable=False)
    method = mapped_column(Integer, ForeignKey("method.id"), nullable=False)  # Honestly, this can just be a string.
    preference = mapped_column(Integer, ForeignKey("preference.id"), nullable=True)
    value = mapped_column(JSON, nullable=False)  # Depends on the method.

Preference

Bases: Base

A model to store user preferences provided by the DM.

Source code in desdeo/api/db_models.py
class Preference(Base):
    """A model to store user preferences provided by the DM."""

    __tablename__ = "preference"
    id: Mapped[int] = mapped_column(primary_key=True, unique=True)
    user = mapped_column(Integer, ForeignKey("user.id"), nullable=False)
    problem = mapped_column(Integer, ForeignKey("problem.id"), nullable=False)
    previous_preference = mapped_column(Integer, ForeignKey("preference.id"), nullable=True)
    method = mapped_column(Integer, ForeignKey("method.id"), nullable=False)
    kind: Mapped[str]  # Depends on the method
    value = mapped_column(JSON, nullable=False)

Problem

Bases: Base

A model to store a problem and its associated data.

Source code in desdeo/api/db_models.py
class Problem(Base):
    """A model to store a problem and its associated data."""

    __tablename__ = "problem"
    id: Mapped[int] = mapped_column(primary_key=True, unique=True)
    owner = mapped_column(Integer, ForeignKey("user.id"), nullable=True)  # Null if problem is public.
    name: Mapped[str] = mapped_column(nullable=False)
    # kind and obj_kind are also in value, but we need them as columns for querying. Maybe?
    kind: Mapped[schema.ProblemKind] = mapped_column(nullable=False)
    obj_kind: Mapped[schema.ObjectiveKind] = mapped_column(nullable=False)
    role_permission: Mapped[list[schema.UserRole]] = mapped_column(ARRAY(Enum(schema.UserRole)), nullable=True)
    # We need some way to tell the API what solver should be used, and this seems like a good place
    # This should match one of the available_solvers in desdeo.tools.utils
    solver: Mapped[schema.Solvers] = mapped_column(nullable=True)
    # Other code assumes these ideals and nadirs are dicts with objective symbols as keys
    presumed_ideal = mapped_column(JSONB, nullable=True)
    presumed_nadir = mapped_column(JSONB, nullable=True)
    # Mapped doesn't work with JSON, so we use JSON directly.
    value = mapped_column(JSON, nullable=False)  # desdeo.problem.schema.Problem

Results

Bases: Base

A model to store the results of a method run.

The results can be partial or complete, depending on the method. For example, NAUTILUS can return ranges instead of solutions. The overlap between the Results and SolutionArchive tables is intentional. Though if you have a better idea, feel free to change it.

Source code in desdeo/api/db_models.py
class Results(Base):
    """A model to store the results of a method run.

    The results can be partial or complete, depending on the method. For example, NAUTILUS can return ranges instead of
    solutions. The overlap between the Results and SolutionArchive tables is intentional. Though if you have a better
    idea, feel free to change it.
    """

    __tablename__ = "results"
    id: Mapped[int] = mapped_column(primary_key=True, unique=True)
    user = mapped_column(Integer, ForeignKey("user.id"), nullable=False)
    problem = mapped_column(Integer, ForeignKey("problem.id"), nullable=False)
    # TODO: The method is temporarily nullable for initial testing. It should be non-nullable.
    method = mapped_column(Integer, ForeignKey("method.id"), nullable=True)
    method_state = mapped_column(Integer, ForeignKey("method_state.id"), nullable=True)
    value = mapped_column(JSON, nullable=False)  # Depends on the method

SolutionArchive

Bases: Base

A model to store a solution archive.

The archive can be used to store the results of a method run. Note that each entry must be a single, complete solution. This is different from the Results table, which can store partial results.

Source code in desdeo/api/db_models.py
class SolutionArchive(Base):
    """A model to store a solution archive.

    The archive can be used to store the results of a method run. Note that each entry must be a single,
    complete solution. This is different from the Results table, which can store partial results.
    """

    __tablename__ = "solution_archive"
    id: Mapped[int] = mapped_column(primary_key=True, unique=True)
    user = mapped_column(Integer, ForeignKey("user.id"), nullable=False)
    problem = mapped_column(Integer, ForeignKey("problem.id"), nullable=False)
    method = mapped_column(Integer, ForeignKey("method.id"), nullable=False)
    preference = mapped_column(Integer, ForeignKey("preference.id"), nullable=True)
    decision_variables = mapped_column(JSONB, nullable=True)
    objectives = mapped_column(ARRAY(FLOAT), nullable=False)
    constraints = mapped_column(ARRAY(FLOAT), nullable=True)
    extra_funcs = mapped_column(ARRAY(FLOAT), nullable=True)
    other_info = mapped_column(
        JSON,
        nullable=True,
    )  # Depends on the method. May include things such as scalarization functions value, etc.
    saved: Mapped[bool] = mapped_column(nullable=False)
    current: Mapped[bool] = mapped_column(nullable=False)
    chosen: Mapped[bool] = mapped_column(nullable=False)

User

Bases: Base

A user with a password, stored problems, role, and user group.

Source code in desdeo/api/db_models.py
class User(Base):
    """A user with a password, stored problems, role, and user group."""

    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True, unique=True)
    username: Mapped[str] = mapped_column(unique=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(nullable=False)
    role: Mapped[schema.UserRole] = mapped_column(nullable=False)
    user_group: Mapped[str] = mapped_column(nullable=True)
    privilages: Mapped[list[schema.UserPrivileges]] = mapped_column(ARRAY(Enum(schema.UserPrivileges)), nullable=False)

    def __repr__(self):
        """Return a string representation of the user (username)."""
        return f"User: ('{self.username}')"
__repr__
__repr__()

Return a string representation of the user (username).

Source code in desdeo/api/db_models.py
def __repr__(self):
    """Return a string representation of the user (username)."""
    return f"User: ('{self.username}')"

UserProblemAccess

Bases: Base

A model to store user's access to problems.

Source code in desdeo/api/db_models.py
class UserProblemAccess(Base):
    """A model to store user's access to problems."""

    __tablename__ = "user_problem_access"
    id: Mapped[int] = mapped_column(primary_key=True, unique=True)
    user_id = mapped_column(Integer, ForeignKey("user.id", ondelete="CASCADE"), nullable=False)
    problem_access: Mapped[int] = mapped_column(Integer, ForeignKey("problem.id"), nullable=False)
    problem = relationship("Problem", foreign_keys=[problem_access], lazy="selectin")

Utopia

Bases: Base

A model to store user specific information relating to Utopia problems.

Source code in desdeo/api/db_models.py
class Utopia(Base):
    """A model to store user specific information relating to Utopia problems."""

    __tablename__ = "utopia"
    id: Mapped[int] = mapped_column(primary_key=True, unique=True)
    problem: Mapped[int] = mapped_column(Integer, ForeignKey("problem.id"), nullable=False)
    map_json: Mapped[str] = mapped_column(nullable=False)
    schedule_dict = mapped_column(JSONB, nullable=False)
    years: Mapped[list[str]] = mapped_column(ARRAY(String), nullable=False)
    stand_id_field: Mapped[str] = mapped_column(String, nullable=False)
    stand_descriptor = mapped_column(JSONB, nullable=True)

Miscellaneous dataclasses used in the database

desdeo.api.schema

Pydantic schemas for the API.

MethodProperties

Bases: Enum

Enum of method properties.

Source code in desdeo/api/schema.py
class MethodProperties(Enum):
    """Enum of method properties."""

    INTERACTIVE = "interactive"
    REFERENCE_POINT = "reference_point"
    CLASSIFICATION = "classification"

Methods

Bases: Enum

Enum of methods.

Source code in desdeo/api/schema.py
class Methods(Enum):
    """Enum of methods."""

    NIMBUS = "nimbus"
    NAUTILUS = "nautilus"
    NAUT_NAVIGATOR = "NAUTILUS navigator"
    NAUTILUSII = "nautilusII"
    RVEA = "RVEA"
    NSGAIII = "NSGAIII"

ObjectiveKind

Bases: Enum

Enum of objective kinds.

Source code in desdeo/api/schema.py
class ObjectiveKind(Enum):
    """Enum of objective kinds."""

    ANALYTICAL = "analytical"
    DATABASED = "databased"
    SIMULATED = "simulated"
    SURROGATE = "surrogate"

ProblemKind

Bases: Enum

Enum of problem kinds.

Source code in desdeo/api/schema.py
class ProblemKind(Enum):
    """Enum of problem kinds."""

    CONTINUOUS = "continuous"
    DISCRETE = "discrete"
    MIXED = "mixed"
    BINARY = "binary"

Solvers

Bases: Enum

Enum of available solvers.

Source code in desdeo/api/schema.py
class Solvers(Enum):
    """Enum of available solvers."""

    # These should match available_solvers in desdeo.tools.utils

    SCIPY_MIN = "scipy_minimize"
    SCIPY_DE = "scipy_de"
    PROXIMAL = "proximal"
    NEVERGRAD = "nevergrad"
    PYOMO_BONMIN = "pyomo_bonmin"
    PYOMO_IPOPT = "pyomo_ipopt"
    PYOMO_GUROBI = "pyomo_gurobi"
    GUROBIPY = "gurobipy"

User

Bases: BaseModel

Model for a user. Temporary.

Source code in desdeo/api/schema.py
class User(BaseModel):
    """Model for a user. Temporary."""

    username: str = Field(description="Username of the user.")
    index: int | None = Field(
        description=(
            "Index of the user in the database. "
            "Supposed to be automatically generated by the database. "
            "So the programmer should not have to worry about it."
        )
    )
    password_hash: str = Field(description="SHA256 Hash of the user's password.")
    role: UserRole = Field(description="Role of the user.")
    privilages: list[UserPrivileges] = Field(description="List of privileges the user has.")
    user_group: str = Field(description="User group of the user. Used for group decision making.")
    # To allows for User to be initialized from database instead of just dicts.
    model_config = ConfigDict(from_attributes=True)

UserPrivileges

Bases: Enum

Enum of user privileges.

Source code in desdeo/api/schema.py
class UserPrivileges(Enum):
    """Enum of user privileges."""

    CREATE_PROBLEMS = "Create problems"
    CREATE_USERS = "Create users"
    ACCESS_ALL_PROBLEMS = "Access all problems"
    EDIT_USERS = "Change user privileges, roles, groups, etc."

UserRole

Bases: Enum

Enum of user roles.

Source code in desdeo/api/schema.py
class UserRole(Enum):
    """Enum of user roles."""

    GUEST = "guest"
    DM = "dm"
    ANALYST = "analyst"

Routes provided by the API

NAUTILUS Navigator

desdeo.api.routers.NAUTILUS_navigator

Endpoints for NAUTILUS Navigator.

InitRequest

Bases: BaseModel

The request to initialize the NAUTILUS Navigator.

Source code in desdeo/api/routers/NAUTILUS_navigator.py
class InitRequest(BaseModel):
    """The request to initialize the NAUTILUS Navigator."""

    problem_id: int = Field(description="The ID of the problem to navigate.")
    """The ID of the problem to navigate."""
    total_steps: int = Field(
        description="The total number of steps in the NAUTILUS Navigator. The default value is 100.", default=100
    )
    "The total number of steps in the NAUTILUS Navigator. The default value is 100."
problem_id class-attribute instance-attribute
problem_id: int = Field(description='The ID of the problem to navigate.')

The ID of the problem to navigate.

total_steps class-attribute instance-attribute
total_steps: int = Field(description='The total number of steps in the NAUTILUS Navigator. The default value is 100.', default=100)

The total number of steps in the NAUTILUS Navigator. The default value is 100.

InitialResponse

Bases: BaseModel

The response from the initial endpoint of NAUTILUS Navigator.

Source code in desdeo/api/routers/NAUTILUS_navigator.py
class InitialResponse(BaseModel):
    """The response from the initial endpoint of NAUTILUS Navigator."""

    objective_symbols: list[str] = Field(description="The symbols of the objectives.")
    objective_long_names: list[str] = Field(description="Long/descriptive names of the objectives.")
    units: list[str] | None = Field(description="The units of the objectives.")
    is_maximized: list[bool] = Field(description="Whether the objectives are to be maximized or minimized.")
    ideal: list[float] = Field(description="The ideal values of the objectives.")
    nadir: list[float] = Field(description="The nadir values of the objectives.")
    total_steps: int = Field(description="The total number of steps in the NAUTILUS Navigator.")
NavigateRequest

Bases: BaseModel

The request to navigate the NAUTILUS Navigator.

Source code in desdeo/api/routers/NAUTILUS_navigator.py
class NavigateRequest(BaseModel):
    """The request to navigate the NAUTILUS Navigator."""

    problem_id: int = Field(description="The ID of the problem to navigate.")
    preference: dict[str, float] = Field(description="The preference of the DM.")
    bounds: dict[str, float] = Field(description="The bounds preference of the DM.")
    go_back_step: int = Field(description="The step index to go back.")
    steps_remaining: int = Field(description="The number of steps remaining. Should be total_steps - go_back_step.")
Response

Bases: BaseModel

The response from most NAUTILUS Navigator endpoints.

Contains information about the full navigation process.

Source code in desdeo/api/routers/NAUTILUS_navigator.py
class Response(BaseModel):
    """The response from most NAUTILUS Navigator endpoints.

    Contains information about the full navigation process.
    """

    objective_symbols: list[str] = Field(description="The symbols of the objectives.")
    objective_long_names: list[str] = Field(description="Long/descriptive names of the objectives.")
    units: list[str] | None = Field(description="The units of the objectives.")
    is_maximized: list[bool] = Field(description="Whether the objectives are to be maximized or minimized.")
    ideal: list[float] = Field(description="The ideal values of the objectives.")
    nadir: list[float] = Field(description="The nadir values of the objectives.")
    lower_bounds: dict[str, list[float]] = Field(description="The lower bounds of the reachable region.")
    upper_bounds: dict[str, list[float]] = Field(description="The upper bounds of the reachable region.")
    preferences: dict[str, list[float]] = Field(description="The preferences used in each step.")
    bounds: dict[str, list[float]] = Field(description="The bounds preference of the DM.")
    total_steps: int = Field(description="The total number of steps in the NAUTILUS Navigator.")
    reachable_solution: dict = Field(description="The solution reached at the end of navigation.")
init_navigator
init_navigator(init_request: InitRequest, user: Annotated[User, Depends(get_current_user)], db: Annotated[Session, Depends(get_db)]) -> InitialResponse

Initialize the NAUTILUS Navigator.

Parameters:

Name Type Description Default
init_request InitRequest

The request to initialize the NAUTILUS Navigator.

required
user Annotated[User, Depends(get_current_user)]

The current user.

required
db Annotated[Session, Depends(get_db)]

The database session.

required

Returns:

Name Type Description
InitialResponse InitialResponse

The initial response from the NAUTILUS Navigator.

Source code in desdeo/api/routers/NAUTILUS_navigator.py
@router.post("/initialize")
def init_navigator(
    init_request: InitRequest,
    user: Annotated[User, Depends(get_current_user)],
    db: Annotated[Session, Depends(get_db)],
) -> InitialResponse:
    """Initialize the NAUTILUS Navigator.

    Args:
        init_request (InitRequest): The request to initialize the NAUTILUS Navigator.
        user (Annotated[User, Depends(get_current_user)]): The current user.
        db (Annotated[Session, Depends(get_db)]): The database session.

    Returns:
        InitialResponse: The initial response from the NAUTILUS Navigator.
    """
    problem_id = init_request.problem_id
    problem = db.query(ProblemInDB).filter(ProblemInDB.id == problem_id).first()

    if problem is None:
        raise HTTPException(status_code=404, detail="Problem not found.")
    if problem.owner != user.index and problem.owner is not None:
        raise HTTPException(status_code=403, detail="Unauthorized to access chosen problem.")
    if problem.value is None:
        raise HTTPException(status_code=500, detail="Problem not found.")
    try:
        problem = Problem.model_validate(problem.value)  # Ignore the mypy error here for now.
    except ValidationError:
        raise HTTPException(status_code=500, detail="Error in parsing the problem.") from ValidationError

    response = navigator_init(problem)

    # Get and delete all Results from previous runs of NAUTILUS Navigator
    results = db.query(Results).filter(Results.problem == problem_id).filter(Results.user == user.index).all()
    for result in results:
        db.delete(result)
    db.commit()

    new_result = Results(
        user=user.index,
        problem=problem_id,
        value=response.model_dump(mode="json"),
    )
    db.add(new_result)
    db.commit()

    return InitialResponse(
        objective_symbols=[obj.symbol for obj in problem.objectives],
        objective_long_names=[obj.name for obj in problem.objectives],
        units=[obj.unit or "" for obj in problem.objectives],  # For unitless objectives, return an empty string
        is_maximized=[obj.maximize for obj in problem.objectives],
        ideal=[obj.ideal for obj in problem.objectives],
        nadir=[obj.nadir for obj in problem.objectives],
        total_steps=init_request.total_steps,
    )
navigate
navigate(request: NavigateRequest, user: Annotated[User, Depends(get_current_user)], db: Annotated[Session, Depends(get_db)]) -> Response

Navigate the NAUTILUS Navigator.

Runs the entire navigation process.

Parameters:

Name Type Description Default
request NavigateRequest

The request to navigate the NAUTILUS Navigator.

required

Raises:

Type Description
HTTPException

description

HTTPException

description

HTTPException

description

HTTPException

description

Returns:

Name Type Description
Response Response

description

Source code in desdeo/api/routers/NAUTILUS_navigator.py
@router.post("/navigate")
def navigate(
    request: NavigateRequest,
    user: Annotated[User, Depends(get_current_user)],
    db: Annotated[Session, Depends(get_db)],
) -> Response:
    """Navigate the NAUTILUS Navigator.

    Runs the entire navigation process.

    Args:
        request (NavigateRequest): The request to navigate the NAUTILUS Navigator.

    Raises:
        HTTPException: _description_
        HTTPException: _description_
        HTTPException: _description_
        HTTPException: _description_

    Returns:
        Response: _description_
    """
    problem_id, preference, go_back_step, steps_remaining, bounds = (
        request.problem_id,
        request.preference,
        request.go_back_step,
        request.steps_remaining,
        request.bounds,
    )
    problem = db.query(ProblemInDB).filter(ProblemInDB.id == problem_id).first()
    if problem is None:
        raise HTTPException(status_code=404, detail="Problem not found.")
    if problem.owner != user.index and problem.owner is not None:
        raise HTTPException(status_code=403, detail="Unauthorized to access chosen problem.")
    try:
        problem = Problem.model_validate(problem.value)  # Ignore the mypy error here for now.
    except ValidationError:
        raise HTTPException(status_code=500, detail="Error in parsing the problem.") from ValidationError

    results = db.query(Results).filter(Results.problem == problem_id).filter(Results.user == user.index).all()
    if not results:
        raise HTTPException(status_code=404, detail="NAUTILUS Navigator not initialized.")

    responses = [NAUTILUS_Response.model_validate(result.value) for result in results]

    responses.append(responses[step_back_index(responses, go_back_step)])

    try:
        new_responses = navigator_all_steps(
            problem,
            steps_remaining=steps_remaining,
            reference_point=preference,
            previous_responses=responses,
            bounds=bounds,
        )
    except IndexError as e:
        raise HTTPException(status_code=400, detail="Possible reason for error: bounds are too restrictive.") from e

    for response in new_responses:
        new_result = Results(
            user=user.index,
            problem=problem_id,
            value=response.model_dump(mode="json"),
        )
        db.add(new_result)
    db.commit()

    responses = [*responses, *new_responses]
    current_path = get_current_path(responses)
    active_responses = [responses[i] for i in current_path]
    lower_bounds = {}
    upper_bounds = {}
    preferences = {}
    bounds = {}
    for obj in problem.objectives:
        lower_bounds[obj.symbol] = [
            response.reachable_bounds["lower_bounds"][obj.symbol] for response in active_responses
        ]
        upper_bounds[obj.symbol] = [
            response.reachable_bounds["upper_bounds"][obj.symbol] for response in active_responses
        ]
        preferences[obj.symbol] = [response.reference_point[obj.symbol] for response in active_responses[1:]]
        bounds[obj.symbol] = [response.bounds[obj.symbol] for response in active_responses[1:]]

    return Response(
        objective_symbols=[obj.symbol for obj in problem.objectives],
        objective_long_names=[obj.name for obj in problem.objectives],
        units=[obj.unit or "" for obj in problem.objectives],
        is_maximized=[obj.maximize for obj in problem.objectives],
        ideal=[obj.ideal for obj in problem.objectives],
        nadir=[obj.nadir for obj in problem.objectives],
        lower_bounds=lower_bounds,
        upper_bounds=upper_bounds,
        bounds=bounds,
        preferences=preferences,
        total_steps=len(active_responses) - 1,
        reachable_solution=active_responses[-1].reachable_solution,
    )