feat: improved validation for writing bookmarks etc. (#365) (#367)

backend/app/schemas/__init__.py

@@ -20,5 +20,6 @@
     SubmittingOrgUpdate,
     VariantPresence,
 )
+from app.schemas.common import RE_HGNCID, RE_SEQVAR, RE_STRUCVAR  # noqa
 from app.schemas.msg import Msg  # noqa
 from app.schemas.user import UserCreate, UserRead, UserUpdate  # noqa

backend/app/schemas/acmgseqvar.py

@@ -3,6 +3,8 @@
 
 from pydantic import BaseModel, ConfigDict
 
+from app.schemas.common import SeqvarName
+
 
 class Presence(str, Enum):
     Present = "Present"
@@ -63,8 +65,8 @@ class AcmgRank(BaseModel):
 
 class AcmgSeqVar(BaseModel):
     user: UUID | None = None
-    seqvar_name: str | None = None
-    acmg_rank: AcmgRank | None = None
+    seqvar_name: SeqvarName
+    acmg_rank: AcmgRank
 
 
 class AcmgSeqVarCreate(AcmgSeqVar):

backend/app/schemas/bookmark.py

@@ -1,7 +1,11 @@
+import re
 from enum import Enum
 from uuid import UUID
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, model_validator
+
+from app.schemas import common
+from app.schemas.common import BookmarkableId
 
 
 class BookmarkTypes(Enum):
@@ -12,8 +16,20 @@ class BookmarkTypes(Enum):
 
 class BookmarkBase(BaseModel):
     user: UUID | None = None
-    obj_type: BookmarkTypes | None = None
-    obj_id: str | None = None
+    obj_type: BookmarkTypes
+    obj_id: BookmarkableId
+
+    @model_validator(mode="after")
+    def check_obj_type_id(self):
+        if self.obj_type == BookmarkTypes.seqvar:
+            assert re.match(common.RE_SEQVAR, self.obj_id), "obj_id is not a valid seqvar"
+        elif self.obj_type == BookmarkTypes.strucvar:
+            assert re.match(common.RE_STRUCVAR, self.obj_id), "obj_id is not a valid strucvar"
+        elif self.obj_type == BookmarkTypes.gene:
+            assert re.match(common.RE_HGNCID, self.obj_id), "obj_id is not a valid HGNC ID"
+        else:
+            assert False, "unknown obj_type"
+        return self
 
 
 class BookmarkCreate(BookmarkBase):

backend/app/schemas/clinvarsub.py

@@ -13,6 +13,7 @@
     SubmissionThreadStatus,
     VariantPresence,
 )
+from app.schemas.common import VarName
 
 
 class SubmittingOrgBase(BaseModel):
@@ -59,7 +60,7 @@ class SubmissionThreadCreate(SubmissionThreadBase):
     model_config = ConfigDict(from_attributes=True)
 
     submittingorg_id: UUID
-    primary_variant_desc: str
+    primary_variant_desc: VarName
 
 
 class SubmissionThreadUpdate(SubmissionThreadBase):
@@ -73,7 +74,7 @@ class SubmissionThreadInDbBase(SubmissionThreadBase):
     created: datetime.datetime
     updated: datetime.datetime
     submittingorg_id: UUID
-    primary_variant_desc: str
+    primary_variant_desc: VarName
 
 
 class SubmissionThreadRead(SubmissionThreadInDbBase):

backend/app/schemas/common.py

@@ -0,0 +1,37 @@
+"""Common schema-related code"""
+
+from typing import TypeAlias
+
+from pydantic import constr
+
+#: Regular expression for sequence variants.
+RE_SEQVAR = r"^(grch37|grch38)-([1-9]|1[0-9]|2[12]|X|Y|MT)-(\d+)-([CGAT]+)-([CGAT]+)$"
+
+#: Regular expression for structurals variants.
+RE_STRUCVAR = r"^(DEL|DUP)-(grch37|grch38)-([1-9]|1[0-9]|2[12]|X|Y|MT)-(\d+)-(\d+)$"
+
+#: Regular expression for HGNC IDs.
+RE_HGNCID = r"^HGNC:(\d+)$"
+
+#: Type for a sequence variant name.
+SeqvarName: TypeAlias = constr(  # type: ignore
+    min_length=1, strip_whitespace=True, pattern=RE_SEQVAR
+)
+
+#: Type for a structural variant name.
+StrucvarName: TypeAlias = constr(  # type: ignore
+    min_length=1, strip_whitespace=True, pattern=RE_STRUCVAR
+)
+
+#: Type for either a sequence or a structural variant name.
+VarName: TypeAlias = constr(  # type: ignore
+    min_length=1, strip_whitespace=True, pattern=f"{RE_SEQVAR}|{RE_STRUCVAR}"
+)
+
+#: Type for a HGNC ID.
+HgncId: TypeAlias = constr(min_length=1, strip_whitespace=True, pattern=RE_HGNCID)  # type: ignore
+
+#: Type for a bookmarkable object.
+BookmarkableId: TypeAlias = constr(  # type: ignore
+    min_length=1, strip_whitespace=True, pattern=f"{RE_SEQVAR}|{RE_STRUCVAR}|{RE_HGNCID}"
+)

backend/tests/api/api_v1/test_acmgseqvar.py

@@ -7,7 +7,7 @@
 
 from app.core.config import settings
 from app.models.user import User
-from tests.conftest import UserChoice
+from tests.conftest import ObjNames, UserChoice
 
 #: Shortcut for regular user.
 REGUL = UserChoice.REGULAR
@@ -20,9 +20,9 @@
 
 
 @pytest.fixture
-def acmgseqvar_post_data() -> dict[str, Any]:
+def acmgseqvar_post_data(obj_names: ObjNames) -> dict[str, Any]:
     return {
-        "seqvar_name": "chr0:123:A:C",
+        "seqvar_name": obj_names.seqvar[0],
         "acmg_rank": {
             "comment": "No comment",
             "criterias": [
@@ -104,14 +104,15 @@ async def test_create_acmgseqvar_invalid_data(
     db_session: AsyncSession,
     client_user: TestClient,
     test_user: User,
+    obj_names: ObjNames,
 ):
     """Test creating a acmgseqvar with invalid data."""
     _ = db_session
     _ = test_user
     # act:
     response = client_user.post(
         f"{settings.API_V1_STR}/acmgseqvar/create",
-        json={"seqvar_name": "chr0:123:A:C", "acmg_rank": {"comment": "No comment"}},
+        json={"seqvar_name": obj_names.seqvar[0], "acmg_rank": {"comment": "No comment"}},
     )
     # assert:
     assert response.status_code == 422
@@ -526,9 +527,9 @@ async def test_get_no_acmgseqvar(
 
 
 @pytest.fixture
-def acmgseqvar_update_data() -> dict[str, Any]:
+def acmgseqvar_update_data(obj_names: ObjNames) -> dict[str, Any]:
     return {
-        "seqvar_name": "chr0:123:A:C",
+        "seqvar_name": obj_names.seqvar[0],
         "acmg_rank": {
             "comment": "Update",
             "criterias": [