Feature/app 47 read corpus type information via admin service

app/api/api_v1/routers/__init__.py

@@ -4,6 +4,7 @@
 from app.api.api_v1.routers.collection import collections_router
 from app.api.api_v1.routers.config import config_router
 from app.api.api_v1.routers.corpus import corpora_router
+from app.api.api_v1.routers.corpus_type import corpus_types_router
 from app.api.api_v1.routers.document import document_router
 from app.api.api_v1.routers.event import event_router
 from app.api.api_v1.routers.family import families_router
@@ -12,6 +13,7 @@
     "analytics_router",
     "auth_router",
     "corpora_router",
+    "corpus_types_router",
     "collections_router",
     "config_router",
     "document_router",

app/api/api_v1/routers/corpus_type.py

@@ -0,0 +1,58 @@
+import logging
+
+from fastapi import APIRouter, HTTPException, Request, status
+
+from app.errors import RepositoryError, ValidationError
+from app.model.corpus_type import CorpusTypeReadDTO
+from app.service import corpus_type as corpus_type_service
+
+corpus_types_router = APIRouter()
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@corpus_types_router.get(
+    "/corpus-types",
+    response_model=list[CorpusTypeReadDTO],
+)
+async def get_all_corpus_types(request: Request) -> list[CorpusTypeReadDTO]:
+    """Retrieve all corpus types.
+
+    :param Request request: Request object.
+    :raises HTTPException: If the corpus type is not found.
+    :return CorpusTypeReadDTO: The requested corpus type.
+    """
+    try:
+        return corpus_type_service.all(request.state.user)
+    except RepositoryError as e:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=e.message
+        )
+
+
+@corpus_types_router.get(
+    "/corpus-types/{corpus_type_name}",
+    response_model=CorpusTypeReadDTO,
+)
+async def get_corpus_type(corpus_type_name: str) -> CorpusTypeReadDTO:
+    """Retrieve a specific corpus type by its name.
+
+    :param str corpus_type_name: The ID of the corpus type to retrieve.
+    :raises HTTPException: If the corpus type is not found.
+    :return CorpusTypeReadDTO: The requested corpus type.
+    """
+    try:
+        corpus_type = corpus_type_service.get(corpus_type_name)
+    except ValidationError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=e.message)
+    except RepositoryError as e:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=e.message
+        )
+
+    if corpus_type is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Corpus type not found: {corpus_type_name}",
+        )
+    return corpus_type

app/main.py

@@ -21,6 +21,7 @@
     collections_router,
     config_router,
     corpora_router,
+    corpus_types_router,
     document_router,
     event_router,
     families_router,
@@ -111,6 +112,14 @@ async def lifespan(app_: FastAPI):
     tags=["corpora"],
     dependencies=[Depends(check_user_auth)],
 )
+
+app.include_router(
+    corpus_types_router,
+    prefix="/api/v1",
+    tags=["corpus-types"],
+    dependencies=[Depends(check_user_auth)],
+)
+
 # Add CORS middleware to allow cross origin requests from any port
 app.add_middleware(
     CORSMiddleware,

app/model/authorisation.py

@@ -20,6 +20,7 @@ class AuthEndpoint(str, enum.Enum):
     EVENT = "EVENTS"
     BULK_IMPORT = "BULK-IMPORT"
     CORPUS = "CORPORA"
+    CORPUS_TYPE = "CORPUS-TYPES"
 
 
 AuthMap = Mapping[AuthEndpoint, Mapping[AuthOperation, AuthAccess]]
@@ -72,4 +73,8 @@ class AuthEndpoint(str, enum.Enum):
         AuthOperation.READ: AuthAccess.SUPER,
         AuthOperation.UPDATE: AuthAccess.SUPER,
     },
+    # Corpus Type
+    AuthEndpoint.CORPUS_TYPE: {
+        AuthOperation.READ: AuthAccess.SUPER,
+    },
 }

app/model/corpus_type.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel
+
+from app.model.general import Json
+
+
+class CorpusTypeReadDTO(BaseModel):
+    """Representation of a Corpus Type."""
+
+    name: str
+    description: str
+    metadata: Json

app/repository/__init__.py

@@ -3,6 +3,7 @@
 import app.repository.collection as collection_repo
 import app.repository.config as config_repo
 import app.repository.corpus as corpus_repo
+import app.repository.corpus_type as corpus_type_repo
 import app.repository.document as document_repo
 import app.repository.event as event_repo
 import app.repository.family as family_repo  # type: ignore
@@ -18,6 +19,7 @@
     "collection_repo",
     "config_repo",
     "corpus_repo",
+    "corpus_type_repo",
     "document_repo",
     "event_repo",
     "family_repo",

app/repository/corpus_type.py

@@ -0,0 +1,66 @@
+import logging
+from typing import Optional, cast
+
+from db_client.models.organisation import CorpusType, Organisation
+from sqlalchemy import asc
+from sqlalchemy.exc import MultipleResultsFound
+from sqlalchemy.orm import Session
+
+from app.errors import RepositoryError
+from app.model.corpus_type import CorpusTypeReadDTO
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def _corpus_type_to_dto(corpus_type: CorpusType) -> CorpusTypeReadDTO:
+    """Convert a CorpusType model to a CorpusTypeReadDTO.
+
+    :param CorpusType corpus_type: The corpus type model.
+    :return CorpusTypeReadDTO: The corresponding DTO.
+    """
+    return CorpusTypeReadDTO(
+        name=str(corpus_type.name),
+        description=str(corpus_type.description),
+        metadata=cast(dict, corpus_type.valid_metadata),
+    )
+
+
+def all(db: Session, org_id: Optional[int]) -> list[CorpusTypeReadDTO]:
+    """Get a list of all corpus types in the database.
+
+    :param db Session: The database connection.
+    :param org_id int: the ID of the organisation the user belongs to
+    :return CorpusTypeReadDTO: The requested corpus type.
+    :raises RepositoryError: If the corpus type is not found.
+    """
+    query = db.query(CorpusType)
+    if org_id is not None:
+        query = query.filter(Organisation.id == org_id)
+
+    result = query.order_by(asc(CorpusType.name)).all()
+
+    if not result:
+        return []
+
+    return [_corpus_type_to_dto(corpus_type) for corpus_type in result]
+
+
+def get(db: Session, corpus_type_name: str) -> Optional[CorpusTypeReadDTO]:
+    """Get a corpus type from the database given a name.
+
+    :param db Session: The database connection.
+    :param str corpus_type_name: The ID of the corpus type to retrieve.
+    :return CorpusTypeReadDTO: The requested corpus type.
+    :raises RepositoryError: If the corpus type is not found.
+    """
+    try:
+        corpus_type = (
+            db.query(CorpusType)
+            .filter(CorpusType.name == corpus_type_name)
+            .one_or_none()
+        )
+        return _corpus_type_to_dto(corpus_type) if corpus_type is not None else None
+
+    except MultipleResultsFound as e:
+        _LOGGER.error(e)
+        raise RepositoryError(e)

app/service/authorisation.py

@@ -71,7 +71,7 @@ def is_authorised(user: UserContext, entity: AuthEndpoint, op: AuthOperation) ->
         return
 
     raise AuthorisationError(
-        f"User {user.email} is not authorised to {op} {_get_article(entity.value)} {entity}"
+        f"User {user.email} is not authorised to {op} {_get_article(entity.value)} {entity.name}"
     )
 
 

app/service/corpus_type.py

@@ -0,0 +1,42 @@
+import logging
+from typing import Optional
+
+from pydantic import ConfigDict, validate_call
+from sqlalchemy.orm import Session
+
+import app.clients.db.session as db_session
+from app.model.corpus_type import CorpusTypeReadDTO
+from app.model.user import UserContext
+from app.repository import corpus_type as corpus_type_repo
+from app.service import app_user
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
+def all(user: UserContext) -> list[CorpusTypeReadDTO]:
+    """
+    Gets the entire list of corpora from the repository.
+
+    :param UserContext user: The current user context.
+    :return list[CorpusReadDTO]: The list of corpora.
+    """
+    with db_session.get_db() as db:
+        org_id = app_user.restrict_entities_to_user_org(user)
+        return corpus_type_repo.all(db, org_id)
+
+
+@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
+def get(
+    corpus_type_name: str, db: Optional[Session] = None
+) -> Optional[CorpusTypeReadDTO]:
+    """Retrieve a corpus type by ID.
+
+    :param str corpus_type_name: The name of the corpus type to retrieve.
+    :return CorpusTypeReadDTO: The requested corpus type.
+    :raises RepositoryError: If there is an error during retrieval.
+    """
+    if db is None:
+        db = db_session.get_db()
+
+    return corpus_type_repo.get(db, corpus_type_name)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "admin_backend"
-version = "2.17.24"
+version = "2.17.25"
 description = ""
 authors = ["CPR-dev-team <[email protected]>"]
 packages = [{ include = "app" }, { include = "tests" }]
@@ -51,7 +51,7 @@ requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
 
 [tool.pytest.ini_options]
-addopts = "-p no:cacheprovider"
+addopts = "--import-mode=importlib"
 env_files = """
     .env.test
     .env

tests/integration_tests/bulk_import/test_bulk_import.py

@@ -402,7 +402,7 @@ def test_bulk_import_admin_non_super(
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
     assert (
-        data["detail"] == "User [email protected] is not authorised to CREATE a BULK-IMPORT"
+        data["detail"] == "User [email protected] is not authorised to CREATE a BULK_IMPORT"
     )
 
 
@@ -416,5 +416,5 @@ def test_bulk_import_non_super_non_admin(
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
     assert (
-        data["detail"] == "User [email protected] is not authorised to CREATE a BULK-IMPORT"
+        data["detail"] == "User [email protected] is not authorised to CREATE a BULK_IMPORT"
     )

tests/integration_tests/bulk_import/test_bulk_import_template.py

@@ -237,7 +237,7 @@ def test_get_template_admin_non_super(
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
     assert (
-        data["detail"] == "User [email protected] is not authorised to READ a BULK-IMPORT"
+        data["detail"] == "User [email protected] is not authorised to READ a BULK_IMPORT"
     )
 
 
@@ -250,4 +250,4 @@ def test_get_template_non_admin_non_super(
     )
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
-    assert data["detail"] == "User [email protected] is not authorised to READ a BULK-IMPORT"
+    assert data["detail"] == "User [email protected] is not authorised to READ a BULK_IMPORT"

tests/integration_tests/corpus/test_all.py

@@ -48,7 +48,7 @@ def test_get_all_corpora_non_super(
     )
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
-    assert data["detail"] == "User [email protected] is not authorised to READ a CORPORA"
+    assert data["detail"] == "User [email protected] is not authorised to READ a CORPUS"
 
 
 def test_get_all_corpora_when_not_authenticated(client: TestClient, data_db: Session):

tests/integration_tests/corpus/test_create.py

@@ -164,7 +164,7 @@ def test_create_corpus_non_admin_non_super(client: TestClient, user_header_token
     )
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
-    assert data["detail"] == "User [email protected] is not authorised to CREATE a CORPORA"
+    assert data["detail"] == "User [email protected] is not authorised to CREATE a CORPUS"
 
 
 def test_create_corpus_admin_non_super(client: TestClient, admin_user_header_token):
@@ -174,7 +174,7 @@ def test_create_corpus_admin_non_super(client: TestClient, admin_user_header_tok
     )
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
-    assert data["detail"] == "User [email protected] is not authorised to CREATE a CORPORA"
+    assert data["detail"] == "User [email protected] is not authorised to CREATE a CORPUS"
 
 
 def test_create_corpus_rollback(

tests/integration_tests/corpus/test_get.py

@@ -43,7 +43,7 @@ def test_get_corpus_non_super(client: TestClient, data_db: Session, user_header_
     )
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
-    assert data["detail"] == "User [email protected] is not authorised to READ a CORPORA"
+    assert data["detail"] == "User [email protected] is not authorised to READ a CORPUS"
 
 
 def test_get_corpus_when_not_authenticated(client: TestClient, data_db: Session):

tests/integration_tests/corpus/test_search.py

@@ -55,7 +55,7 @@ def test_search_corpus_non_super(
     )
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
-    assert data["detail"] == "User [email protected] is not authorised to READ a CORPORA"
+    assert data["detail"] == "User [email protected] is not authorised to READ a CORPUS"
 
 
 def test_search_corpus_when_not_authorised(client: TestClient, data_db: Session):

tests/integration_tests/corpus/test_update.py

@@ -165,7 +165,7 @@ def test_update_corpus_non_super_non_admin(client: TestClient, user_header_token
     )
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
-    assert data["detail"] == "User [email protected] is not authorised to UPDATE a CORPORA"
+    assert data["detail"] == "User [email protected] is not authorised to UPDATE a CORPUS"
 
 
 def test_update_corpus_non_super_admin(client: TestClient, admin_user_header_token):
@@ -177,7 +177,7 @@ def test_update_corpus_non_super_admin(client: TestClient, admin_user_header_tok
     )
     assert response.status_code == status.HTTP_403_FORBIDDEN
     data = response.json()
-    assert data["detail"] == "User [email protected] is not authorised to UPDATE a CORPORA"
+    assert data["detail"] == "User [email protected] is not authorised to UPDATE a CORPUS"
 
 
 def test_update_corpus_idempotent(