Add processes endpoint

aiida_restapi/main.py

@@ -3,11 +3,12 @@
 from fastapi import FastAPI
 
 from aiida_restapi.graphql import main
-from aiida_restapi.routers import auth, computers, groups, users
+from aiida_restapi.routers import auth, computers, groups, process, users
 
 app = FastAPI()
 app.include_router(auth.router)
 app.include_router(computers.router)
 app.include_router(groups.router)
 app.include_router(users.router)
+app.include_router(process.router)
 app.add_route("/graphql", main.app, name="graphql")

aiida_restapi/models.py

@@ -7,7 +7,8 @@
 # pylint: disable=too-few-public-methods
 
 from datetime import datetime
-from typing import ClassVar, List, Optional, Type, TypeVar
+from typing import ClassVar, Dict, List, Optional, Type, TypeVar
+from uuid import UUID
 
 from aiida import orm
 from pydantic import BaseModel, Field
@@ -129,3 +130,26 @@ class Group(AiidaModel):
     type_string: Optional[str] = Field(description="Type of the group")
     user_id: Optional[str] = Field(description="Id of the user that created the node.")
     description: Optional[str] = Field(description="Short description of the group.")
+
+
+class Process(AiidaModel):
+    """AiiDA Process Model"""
+
+    _orm_entity = orm.ProcessNode
+
+    id: Optional[int] = Field(description="Unique id (pk)")
+    uuid: Optional[UUID] = Field(description="Unique uuid")
+    node_type: Optional[str] = Field(description="Node type")
+    process_type: Optional[str] = Field(description="Process type")
+    label: str = Field(description="Label of node")
+    description: Optional[str] = Field(description="Description of node")
+    ctime: Optional[datetime] = Field(description="Creation time")
+    mtime: Optional[datetime] = Field(description="Last modification time")
+    user_id: Optional[int] = Field(description="Created by user id (pk)")
+    dbcomputer_id: Optional[int] = Field(description="Associated computer id (pk)")
+    attributes: Optional[Dict] = Field(
+        description="Variable attributes of the node",
+    )
+    extras: Optional[Dict] = Field(
+        description="Variable extras (unsealed) of the node",
+    )

aiida_restapi/routers/process.py

@@ -0,0 +1,40 @@
+# -*- coding: utf-8 -*-
+"""Declaration of FastAPI application."""
+
+from typing import List, Optional
+
+from aiida import orm
+from aiida.cmdline.utils.decorators import with_dbenv
+from aiida.orm.querybuilder import QueryBuilder
+from fastapi import APIRouter
+
+from aiida_restapi.models import Process
+
+router = APIRouter()
+
+
+@router.get("/processes", response_model=List[Process])
+@with_dbenv()
+async def read_processes() -> List[Process]:
+    """Get list of all processes"""
+
+    return Process.get_entities()
+
+
+@router.get("/processes/projectable_properties", response_model=List[str])
+async def get_processes_projectable_properties() -> List[str]:
+    """Get projectable properties for processes endpoint"""
+
+    return Process.get_projectable_properties()
+
+
+@router.get("/processes/{proc_id}", response_model=Process)
+@with_dbenv()
+async def read_process(proc_id: int) -> Optional[Process]:
+    """Get process by id."""
+    qbobj = QueryBuilder()
+    qbobj.append(
+        orm.ProcessNode, filters={"id": proc_id}, project=["**"], tag="process"
+    ).limit(1)
+
+    return qbobj.dict()[0]["process"]

conftest.py

@@ -2,6 +2,8 @@
 """pytest fixtures for simplified testing."""
 import pytest
 from aiida import orm
+from aiida.engine import ProcessState
+from aiida.orm import WorkChainNode, WorkFunctionNode
 from fastapi.testclient import TestClient
 
 from aiida_restapi import app, config
@@ -53,6 +55,44 @@ def default_computers():
     return [comp_1.id, comp_2.id]
 
 
+@pytest.fixture(scope="function")
+def default_process():
+    """Populate database with some process"""
+    calcs = []
+    process_label = "SomeDummyWorkFunctionNode"
+
+    # Create 6 WorkFunctionNodes and WorkChainNodes (one for each ProcessState)
+    for state in ProcessState:
+
+        calc = WorkFunctionNode()
+        calc.set_process_state(state)
+
+        # Set the WorkFunctionNode as successful
+        if state == ProcessState.FINISHED:
+            calc.set_exit_status(0)
+
+        # Give a `process_label` to the `WorkFunctionNodes` so the `--process-label` option can be tested
+        calc.set_attribute("process_label", process_label)
+
+        calc.store()
+        calcs.append(calc.id)
+
+        calc = WorkChainNode()
+        calc.set_process_state(state)
+
+        # Set the WorkChainNode as failed
+        if state == ProcessState.FINISHED:
+            calc.set_exit_status(1)
+
+        # Set the waiting work chain as paused as well
+        if state == ProcessState.WAITING:
+            calc.pause()
+
+        calc.store()
+        calcs.append(calc.id)
+    return calcs
+
+
 @pytest.fixture(scope="function")
 def default_groups():
     """Populate database with some groups."""

tests/test_graphql/test_full/test_full.yml

@@ -1,4 +1,4 @@
 data:
-  aiidaVersion: 1.6.3
+  aiidaVersion: 1.6.4
   node:
     label: node 1