diff --git a/backend/app/app/api/endpoints/keggs.py b/backend/app/app/api/endpoints/keggs.py
new file mode 100644
index 0000000000000000000000000000000000000000..eceda53cf75cad826c103e3bed266b0ea91e6bd7
--- /dev/null
+++ b/backend/app/app/api/endpoints/keggs.py
@@ -0,0 +1,61 @@
+from typing import List
+
+from fastapi import Depends, APIRouter, HTTPException
+from sqlalchemy.exc import NoResultFound, IntegrityError
+from sqlmodel import Session
+
+from app.db import get_session
+from app.core.schemas.entities.kegg import KeggRead, KeggUpdate, KeggCreate
+from app.core.use_cases.crud.kegg import CrudKeggUseCase
+
+
+router = APIRouter()
+
+
+@router.get("/", response_model=List[KeggRead])
+async def get_keggs(session: Session = Depends(get_session)):
+ use_case = CrudKeggUseCase()
+ keggs = use_case.get_all(session=session)
+ return keggs
+
+
+@router.get("/{kegg_id}", response_model=KeggRead)
+async def get_kegg(kegg_id: str, session: Session = Depends(get_session)):
+ use_case = CrudKeggUseCase()
+ try:
+ kegg = use_case.get_kegg(kegg_id, session=session)
+ except NoResultFound:
+ raise HTTPException(status_code=404, detail=f"Kegg [{kegg_id}] not found.")
+ return kegg
+
+
+@router.post("/", response_model=KeggRead)
+async def create_kegg(kegg: KeggCreate, session: Session = Depends(get_session)):
+ use_case = CrudKeggUseCase()
+ try:
+ kegg = use_case.create_kegg(kegg, session=session)
+ except IntegrityError:
+ raise HTTPException(
+ status_code=403, detail=f"Kegg [{kegg.name}] already exists."
+ )
+ return kegg
+
+
+@router.put("/", response_model=KeggUpdate)
+async def update_kegg(kegg: KeggCreate, session: Session = Depends(get_session)):
+ use_case = CrudKeggUseCase()
+ try:
+ kegg = use_case.update_kegg(kegg, session=session)
+ except NoResultFound:
+ raise HTTPException(status_code=404, detail=f"Kegg [{kegg.name}] not found.")
+ return kegg
+
+
+@router.delete("/{kegg_id}")
+async def delete_kegg(kegg_id: str, session: Session = Depends(get_session)):
+ use_case = CrudKeggUseCase()
+ try:
+ use_case.delete_kegg(kegg_id, session=session)
+ except NoResultFound:
+ raise HTTPException(status_code=404, detail=f"Kegg [{kegg_id}] not found.")
+ return {"deleted": True}
diff --git a/backend/app/app/api/router.py b/backend/app/app/api/router.py
index 3dfd0e244c3fa849c104c940737d2b013fbaee93..2f7a5a750dd414c10b8687beeec7be58941c165f 100644
--- a/backend/app/app/api/router.py
+++ b/backend/app/app/api/router.py
@@ -1,6 +1,6 @@
from fastapi import APIRouter
-from app.api.endpoints import catalogs, experiments, genes
+from app.api.endpoints import catalogs, experiments, genes, keggs
api_router = APIRouter()
api_router.include_router(genes.router, prefix="/genes", tags=["Genes"])
@@ -8,3 +8,4 @@ api_router.include_router(catalogs.router, prefix="/catalogs", tags=["Catalogs"]
api_router.include_router(
experiments.router, prefix="/experiments", tags=["Experiments"]
)
+api_router.include_router(keggs.router, prefix="/keggs", tags=["Keggs"])
diff --git a/backend/app/app/core/models/kegg.py b/backend/app/app/core/models/kegg.py
index 12ebe7ec7bcf7d4723ebe4301d8920071a4c82fd..0aa19e08c3a27862afdb47471fe2408964416a65 100644
--- a/backend/app/app/core/models/kegg.py
+++ b/backend/app/app/core/models/kegg.py
@@ -1,11 +1,11 @@
from typing import List
from slugify import slugify
-from sqlmodel import Field, Relationship, SQLModel
+from sqlmodel import Field, Relationship, SQLModel, VARCHAR, Column
class KeggBase(SQLModel):
- kegg_id: str
+ kegg_id: str = Field(sa_column=Column("kegg_id", VARCHAR, unique=True))
name: str = Field(index=False)
@property
diff --git a/backend/app/app/core/repositories/kegg.py b/backend/app/app/core/repositories/kegg.py
new file mode 100644
index 0000000000000000000000000000000000000000..c480d167bde4b9682ed37fc6e05fcaae26ceb15e
--- /dev/null
+++ b/backend/app/app/core/repositories/kegg.py
@@ -0,0 +1,64 @@
+import abc
+from typing import List
+
+from sqlmodel import Session, select
+
+from app.core.models.kegg import Kegg
+from app.core.schemas.entities.kegg import KeggCreate, KeggUpdate
+
+
+class KeggsRepo(abc.ABC):
+ @abc.abstractmethod
+ def get(self, kegg_name: str) -> Kegg:
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def get_all(self) -> List[Kegg]:
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def create(self, kegg_input: KeggCreate) -> Kegg:
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def update(self, kegg_input: KeggUpdate) -> Kegg:
+ raise NotImplementedError
+
+ @abc.abstractmethod
+ def delete(self, kegg_name: str):
+ raise NotImplementedError
+
+
+class SqlModelKeggsRepo(KeggsRepo):
+ def get(self, kegg_id, session: Session) -> Kegg:
+ """Retrieve one kegg from name."""
+ statement = select(Kegg).where(Kegg.kegg_id == kegg_id)
+ return session.exec(statement).one()
+
+ def get_all(self, session: Session) -> List[Kegg]:
+ """Retrieve all keggs."""
+ keggs = session.exec(select(Kegg)).all()
+ return keggs
+
+ def create(self, kegg_input: KeggCreate, session: Session) -> Kegg:
+ """Create a kegg entry."""
+ kegg = Kegg(kegg_id=kegg_input.kegg_id, name=kegg_input.name)
+ session.add(kegg)
+ session.commit()
+ session.refresh(kegg)
+ return kegg
+
+ def update(self, kegg_input: KeggUpdate, session: Session) -> Kegg:
+ """Update a kegg entry."""
+ kegg = self.get(kegg_input.kegg_id, session)
+ for k, v in kegg_input.dict().items():
+ setattr(kegg, k, v)
+ session.add(kegg)
+ session.commit()
+ session.refresh(kegg)
+ return kegg
+
+ def delete(self, kegg_id: str, session: Session):
+ kegg = self.get(kegg_id, session)
+ session.delete(kegg)
+ session.commit()
diff --git a/backend/app/app/core/schemas/entities/kegg.py b/backend/app/app/core/schemas/entities/kegg.py
index bbd565c98e8ccfdb4bb6b6139b4608b606bbd0e5..953ae670178a607a498b04c940aa52da4ca12e28 100644
--- a/backend/app/app/core/schemas/entities/kegg.py
+++ b/backend/app/app/core/schemas/entities/kegg.py
@@ -7,3 +7,7 @@ class KeggRead(KeggBase):
class KeggCreate(KeggBase):
pass
+
+
+class KeggUpdate(KeggBase):
+ pass
diff --git a/backend/app/app/core/use_cases/crud/kegg.py b/backend/app/app/core/use_cases/crud/kegg.py
new file mode 100644
index 0000000000000000000000000000000000000000..0fc173f1fdfa636cedd58c436c2b701c73a1e5f1
--- /dev/null
+++ b/backend/app/app/core/use_cases/crud/kegg.py
@@ -0,0 +1,28 @@
+from typing import List
+
+import inject
+
+from app.core.models.kegg import Kegg
+from app.core.schemas.entities.kegg import KeggCreate, KeggUpdate
+from app.core.repositories.kegg import KeggsRepo
+
+
+class CrudKeggUseCase:
+ @inject.autoparams("keggs_repo")
+ def __init__(self, keggs_repo: KeggsRepo):
+ self._keggs_repo = keggs_repo
+
+ def create_kegg(self, kegg_input: KeggCreate, **kwargs) -> Kegg:
+ return self._keggs_repo.create(kegg_input, **kwargs)
+
+ def update_kegg(self, kegg_input: KeggUpdate, **kwargs) -> Kegg:
+ return self._keggs_repo.update(kegg_input, **kwargs)
+
+ def delete_kegg(self, kegg_name: str, **kwargs):
+ return self._keggs_repo.delete(kegg_name, **kwargs)
+
+ def get_kegg(self, kegg_id: str, **kwargs) -> Kegg:
+ return self._keggs_repo.get(kegg_id, **kwargs)
+
+ def get_all(self, **kwargs) -> List[Kegg]:
+ return self._keggs_repo.get_all(**kwargs)
diff --git a/backend/app/app/dependency_injections.py b/backend/app/app/dependency_injections.py
index 2a409860856c2b130c4c9691b42f631d747c15a9..10f4c827a80191219e29a1854c1555741765190f 100644
--- a/backend/app/app/dependency_injections.py
+++ b/backend/app/app/dependency_injections.py
@@ -1,11 +1,17 @@
import inject
from app.core.repositories.catalog import CatalogsRepo, SqlModelCatalogsRepo
+from app.core.repositories.kegg import KeggsRepo, SqlModelKeggsRepo
def di_configuration(binder):
binder.bind(CatalogsRepo, SqlModelCatalogsRepo())
+ binder.bind(KeggsRepo, SqlModelKeggsRepo())
def run_di():
inject.configure(di_configuration)
+
+
+def clear_di():
+ inject.clear()
diff --git a/backend/app/tests/api/endpoints/base_api_test.py b/backend/app/tests/api/endpoints/base_api_test.py
new file mode 100644
index 0000000000000000000000000000000000000000..322fdd31c6e5cfc1dd8fe98ee55c1da1dd1a7a3c
--- /dev/null
+++ b/backend/app/tests/api/endpoints/base_api_test.py
@@ -0,0 +1,44 @@
+import unittest
+
+from sqlmodel import create_engine, Session, SQLModel
+from sqlmodel.pool import StaticPool
+from fastapi.testclient import TestClient
+
+from app.db import get_session
+from app.dependency_injections import run_di, clear_di
+from app.main import app
+
+
+class BaseApiTests(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls) -> None:
+ # Dependency injections
+ run_di()
+
+ cls.engine = create_engine(
+ "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
+ )
+ SQLModel.metadata.create_all(cls.engine)
+ with Session(cls.engine) as session:
+ cls._create_test_db(session)
+
+ @classmethod
+ def _create_test_db(cls, session: Session):
+ pass
+
+ def setUp(self) -> None:
+ self.session = Session(self.engine)
+
+ def get_session_override():
+ return self.session
+
+ app.dependency_overrides[get_session] = get_session_override
+ self.client = TestClient(app)
+
+ def tearDown(self) -> None:
+ app.dependency_overrides.clear()
+ self.session.close()
+
+ @classmethod
+ def tearDownClass(cls) -> None:
+ clear_di()
diff --git a/backend/app/tests/api/endpoints/test_catalogs.py b/backend/app/tests/api/endpoints/test_catalogs.py
index 4bb7f4cf61b37d48114d19f21ef90a64713851b4..98840a5da2162920814e9e52a82d31b9da7f9f1c 100644
--- a/backend/app/tests/api/endpoints/test_catalogs.py
+++ b/backend/app/tests/api/endpoints/test_catalogs.py
@@ -1,14 +1,9 @@
-import unittest
+from sqlmodel import Session, select
-from sqlmodel import create_engine, Session, SQLModel, select
-from sqlmodel.pool import StaticPool
-from fastapi.testclient import TestClient
-
-from app.db import get_session
-from app.dependency_injections import run_di
-from app.main import app
from app.core.models.catalog import Catalog
+from .base_api_test import BaseApiTests
+
def create_test_db(session: Session):
test_catalog = Catalog(name="test-catalog")
@@ -16,31 +11,12 @@ def create_test_db(session: Session):
session.commit()
-class TestCatalogs(unittest.TestCase):
+class TestCatalogs(BaseApiTests):
@classmethod
- def setUpClass(cls) -> None:
- # Dependency injections
- run_di()
-
- cls.engine = create_engine(
- "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
- )
- SQLModel.metadata.create_all(cls.engine)
- with Session(cls.engine) as session:
- create_test_db(session)
-
- def setUp(self) -> None:
- self.session = Session(self.engine)
-
- def get_session_override():
- return self.session
-
- app.dependency_overrides[get_session] = get_session_override
- self.client = TestClient(app)
-
- def tearDown(self) -> None:
- app.dependency_overrides.clear()
- self.session.close()
+ def _create_test_db(cls, session: Session):
+ test_catalog = Catalog(name="test-catalog")
+ session.add(test_catalog)
+ session.commit()
def test_get_catalog_non_existing(self):
# When
diff --git a/backend/app/tests/api/endpoints/test_keggs.py b/backend/app/tests/api/endpoints/test_keggs.py
new file mode 100644
index 0000000000000000000000000000000000000000..2b2ca84066e2f3f88f29126ff3e8e8faf5640a01
--- /dev/null
+++ b/backend/app/tests/api/endpoints/test_keggs.py
@@ -0,0 +1,116 @@
+from sqlmodel import Session, select
+
+from app.core.models.kegg import Kegg
+
+from .base_api_test import BaseApiTests
+
+
+def create_test_db(session: Session):
+ test_kegg = Kegg(name="test-kegg")
+ session.add(test_kegg)
+ session.commit()
+
+
+class TestKeggs(BaseApiTests):
+ @classmethod
+ def _create_test_db(cls, session: Session):
+ test_kegg = Kegg(kegg_id="k1", name="test-kegg")
+ session.add(test_kegg)
+ session.commit()
+
+ def test_get_kegg_non_existing(self):
+ # When
+ response = self.client.get("/api/keggs/non-existing")
+ # Then
+ self.assertEqual(response.status_code, 404)
+
+ def test_get_kegg_existing(self):
+ # Given
+ expected_id = "k1"
+ expected_data = {"kegg_id": expected_id, "name": "test-kegg"}
+ # When
+ response = self.client.get(f"/api/keggs/{expected_id}")
+ # Then
+ self.assertEqual(response.status_code, 200)
+ self.assertDictEqual(response.json(), expected_data)
+
+ def test_get_all_keggs(self):
+ # Given
+ expected_data = [{"kegg_id": "k1", "name": "test-kegg"}]
+ # When
+ response = self.client.get(f"/api/keggs/")
+ # Then
+ self.assertEqual(response.status_code, 200)
+ self.assertListEqual(response.json(), expected_data)
+
+ def test_create_kegg(self):
+ # Given
+ kegg_id = "k9"
+ kegg_name = "created_kegg"
+ json_input = {"kegg_id": kegg_id, "name": kegg_name}
+ expected_data = {"kegg_id": kegg_id, "name": kegg_name}
+ # When
+ response = self.client.post("/api/keggs/", json=json_input)
+ # Then
+ self.assertEqual(response.status_code, 200)
+ self.assertDictEqual(response.json(), expected_data)
+ # Finally delete item
+ cat = self.session.exec(select(Kegg).where(Kegg.kegg_id == kegg_id)).one()
+ self.session.delete(cat)
+ self.session.commit()
+
+ def test_create_existing_kegg(self):
+ # Given
+ kegg_id = "k1"
+ kegg_name = "test-kegg"
+ json_input = {"kegg_id": kegg_id, "name": kegg_name}
+ # When
+ response = self.client.post("/api/keggs/", json=json_input)
+ # Then
+ self.assertEqual(response.status_code, 403)
+
+ def test_delete_kegg_non_existing(self):
+ # When
+ response = self.client.delete(f"/api/keggs/non-existing")
+ # Then
+ self.assertEqual(response.status_code, 404)
+
+ def test_delete_kegg(self):
+ # Given
+ kegg_id = "k5"
+ name_to_delete = "delete-me"
+ # - create kegg to delete
+ kegg_to_delete = Kegg(kegg_id=kegg_id, name=name_to_delete)
+ self.session.add(kegg_to_delete)
+ self.session.commit()
+ # When
+ response = self.client.delete(f"/api/keggs/{kegg_id}")
+ # Then
+ self.assertEqual(response.status_code, 200)
+
+ def test_update_kegg_non_existing(self):
+ # When
+ json_input = {"kegg_id": "i-dont-exist", "name": "12345"}
+ response = self.client.put(f"/api/keggs/", json=json_input)
+ # Then
+ self.assertEqual(response.status_code, 404)
+
+ def test_update_kegg(self):
+ # Given
+ kegg_id = "k2"
+ name_to_update = "update-me"
+ new_name = "new-name"
+ # - create kegg to update
+ kegg_to_update = Kegg(kegg_id=kegg_id, name=name_to_update)
+ self.session.add(kegg_to_update)
+ self.session.commit()
+ json_input = {"kegg_id": kegg_id, "name": new_name}
+ expected_data = json_input
+ # When
+ response = self.client.put(f"/api/keggs/", json=json_input)
+ # Then
+ self.assertEqual(response.status_code, 200)
+ self.assertDictEqual(response.json(), expected_data)
+ # Finally
+ self.session.delete(kegg_to_update)
+ self.session.commit()
diff --git a/backend/app/tests/core/use_cases/crud/test_kegg.py b/backend/app/tests/core/use_cases/crud/test_kegg.py
new file mode 100644
index 0000000000000000000000000000000000000000..ffa2c5f8f306acce15a2d7b12b627a7b3f92e5ec
--- /dev/null
+++ b/backend/app/tests/core/use_cases/crud/test_kegg.py
@@ -0,0 +1,76 @@
+import unittest
+
+from app.core.use_cases.crud.kegg import CrudKeggUseCase
+from app.core.schemas.entities.kegg import KeggCreate, KeggUpdate
+
+GET_ACTION = "Returning kegg"
+GET_ALL_ACTION = "Returning all keggs"
+CREATE_ACTION = "Creating kegg"
+UPDATE_ACTION = "Updating kegg"
+DELETE_ACTION = "Deleting kegg"
+
+
+class MockKeggsRepo:
+ def get(self, kegg_id: str):
+ return GET_ACTION
+
+ def get_all(self):
+ return GET_ALL_ACTION
+
+ def create(self, kegg_input: KeggCreate):
+ return CREATE_ACTION
+
+ def update(self, kegg_input: KeggUpdate):
+ return UPDATE_ACTION
+
+ def delete(self, kegg_id: str):
+ return DELETE_ACTION
+
+
+class TestCrudKeggUseCase(unittest.TestCase):
+ def setUp(self) -> None:
+ self.use_case_test = CrudKeggUseCase(keggs_repo=MockKeggsRepo())
+
+ def test_create_kegg(self):
+ # Given
+ kegg_test = KeggCreate(kegg_id="k01", name="test-kegg")
+ expected_output = CREATE_ACTION
+ # When
+ test_output = self.use_case_test.create_kegg(kegg_test)
+ # Then
+ self.assertEqual(test_output, expected_output)
+
+ def test_update_kegg(self):
+ # Given
+ kegg_test = KeggCreate(kegg_id="k01", name="test-kegg")
+ expected_output = UPDATE_ACTION
+ # When
+ test_output = self.use_case_test.update_kegg(kegg_test)
+ # Then
+ self.assertEqual(test_output, expected_output)
+
+ def test_delete_kegg(self):
+ # Given
+ kegg_id_test = "test-kegg"
+ expected_output = DELETE_ACTION
+ # When
+ test_output = self.use_case_test.delete_kegg(kegg_id_test)
+ # Then
+ self.assertEqual(test_output, expected_output)
+
+ def test_get_kegg(self):
+ # Given
+ kegg_id_test = "test-kegg"
+ expected_output = GET_ACTION
+ # When
+ test_output = self.use_case_test.get_kegg(kegg_id_test)
+ # Then
+ self.assertEqual(test_output, expected_output)
+
+ def test_get_all(self):
+ # Given
+ expected_output = GET_ALL_ACTION
+ # When
+ output = self.use_case_test.get_all()
+ # Then
+ self.assertEqual(output, expected_output)