Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 43 additions & 0 deletions packages/opencloning-db/src/opencloning_db/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,14 @@
Enum,
ForeignKey,
ForeignKeyConstraint,
Index,
Integer,
JSON,
Table,
UniqueConstraint,
event,
select,
text,
)

from sqlalchemy.sql import func
Expand Down Expand Up @@ -202,6 +205,15 @@ class Tag(Base):

class InputEntity(Base):
__tablename__ = 'input_entity'
__table_args__ = (
Index(
'uq_input_entity_template_sequence_workspace_name',
'workspace_id',
text('lower(name)'),
unique=True,
postgresql_where=text("type = 'template_sequence'"),
),
)
Comment on lines +208 to +216

id: Mapped[int] = mapped_column(primary_key=True)
workspace_id: Mapped[int] = mapped_column(ForeignKey('workspace.id'), nullable=False)
Expand Down Expand Up @@ -364,6 +376,37 @@ def to_pydantic_sequence(self) -> opencloning_models.TemplateSequence:
)


def template_sequence_name_taken(
session: SASession,
*,
workspace_id: int,
name: str,
exclude_id: int | None = None,
) -> bool:
"""Return whether another template sequence in the workspace already has this name (case-insensitive)."""
stmt = select(TemplateSequence.id).where(
TemplateSequence.workspace_id == workspace_id,
func.lower(TemplateSequence.name) == name.lower(),
)
if exclude_id is not None:
stmt = stmt.where(TemplateSequence.id != exclude_id)
return session.scalar(stmt.limit(1)) is not None


def assert_template_sequence_name_available(
session: SASession,
*,
workspace_id: int,
name: str,
exclude_id: int | None = None,
) -> None:
if template_sequence_name_taken(session, workspace_id=workspace_id, name=name, exclude_id=exclude_id):
raise HTTPException(
status_code=409,
detail=f"Template sequence '{name}' already exists in this workspace",
)


class Primer(InputEntity):
__tablename__ = 'primer'

Expand Down
20 changes: 19 additions & 1 deletion packages/opencloning-db/src/opencloning_db/routers/sequences.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,10 @@
Tag,
SequenceSample,
SourceInput,
TemplateSequence,
User,
WorkspaceRole,
assert_template_sequence_name_available,
require_real_sequence,
)
from fastapi_pagination import Page
Expand Down Expand Up @@ -165,6 +167,13 @@ def patch_sequence(
)

if body.name is not None:
if isinstance(db_sequence, TemplateSequence):
assert_template_sequence_name_available(
session,
workspace_id=workspace_id,
name=body.name,
exclude_id=db_sequence.id,
)
db_sequence.name = body.name

if body.sequence_type is not None:
Expand All @@ -180,7 +189,16 @@ def patch_sequence(
)
db_sequence.sequence_type = body.sequence_type

session.commit()
try:
session.commit()
except IntegrityError:
session.rollback()
if body.name is not None and isinstance(db_sequence, TemplateSequence):
raise HTTPException(
status_code=409,
detail=f"Template sequence '{body.name}' already exists in this workspace",
) from None
raise
session.refresh(db_sequence)
return sequence_ref(db_sequence)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,11 @@

from typing import Annotated

from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.exc import IntegrityError

from opencloning_db.apimodels import SequenceRef, TemplateSequenceCreate, sequence_ref
from opencloning_db.models import TemplateSequence
from opencloning_db.models import TemplateSequence, assert_template_sequence_name_available
from opencloning_db.workspace_deps import WorkspaceContext, get_editor_workspace_ctx

router = APIRouter(tags=['template_sequences'])
Expand All @@ -16,13 +17,21 @@ def post_template_sequence(
ctx: Annotated[WorkspaceContext, Depends(get_editor_workspace_ctx)],
body: TemplateSequenceCreate,
):
_, session, _ = ctx.destructure()
_, session, workspace_id = ctx.destructure()
assert_template_sequence_name_available(session, workspace_id=workspace_id, name=body.name)
template_sequence = TemplateSequence.from_create(
name=body.name,
sequence_type=body.sequence_type,
ctx=ctx,
)
session.add(template_sequence)
session.commit()
try:
session.commit()
except IntegrityError:
session.rollback()
raise HTTPException(
status_code=409,
detail=f"Template sequence '{body.name}' already exists in this workspace",
) from None
session.refresh(template_sequence)
return sequence_ref(template_sequence)
81 changes: 81 additions & 0 deletions packages/opencloning-db/tests/test_template_sequences.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,87 @@ def test_post_template_sequence_persists_template_subtype(template_sequences_cli
assert stored.name == 'Template Allele'


def test_post_template_sequence_duplicate_name_409(template_sequences_client):
c = template_sequences_client['client']
headers = workspace_headers(template_sequences_client['token_owner_w1'], template_sequences_client['w1'])
first = c.post(
'/template_sequences',
headers=headers,
json={'name': 'Shared Template', 'sequence_type': 'allele'},
)
assert first.status_code == 200

second = c.post(
'/template_sequences',
headers=headers,
json={'name': 'Shared Template', 'sequence_type': 'plasmid'},
)
assert second.status_code == 409
assert 'already exists' in second.json()['detail']


def test_post_template_sequence_duplicate_name_case_insensitive_409(template_sequences_client):
c = template_sequences_client['client']
headers = workspace_headers(template_sequences_client['token_owner_w1'], template_sequences_client['w1'])
first = c.post(
'/template_sequences',
headers=headers,
json={'name': 'Case Template', 'sequence_type': 'allele'},
)
assert first.status_code == 200

second = c.post(
'/template_sequences',
headers=headers,
json={'name': 'case template', 'sequence_type': 'allele'},
)
assert second.status_code == 409
assert 'already exists' in second.json()['detail']


def test_patch_template_sequence_duplicate_name_409(template_sequences_client):
c = template_sequences_client['client']
headers = workspace_headers(template_sequences_client['token_owner_w1'], template_sequences_client['w1'])
first = c.post(
'/template_sequences',
headers=headers,
json={'name': 'Rename Target', 'sequence_type': 'allele'},
)
assert first.status_code == 200
other = c.post(
'/template_sequences',
headers=headers,
json={'name': 'Other Template', 'sequence_type': 'allele'},
)
assert other.status_code == 200

patch = c.patch(
f"/sequences/{first.json()['id']}",
headers=headers,
json={'name': 'other template'},
)
assert patch.status_code == 409
assert 'already exists' in patch.json()['detail']


def test_post_template_sequence_same_name_different_workspace_ok(template_sequences_client):
c = template_sequences_client['client']
w1_headers = workspace_headers(template_sequences_client['token_owner_w1'], template_sequences_client['w1'])
w2_headers = workspace_headers(template_sequences_client['token_owner_w2'], template_sequences_client['w2'])
w1 = c.post(
'/template_sequences',
headers=w1_headers,
json={'name': 'Cross Workspace', 'sequence_type': 'allele'},
)
assert w1.status_code == 200
w2 = c.post(
'/template_sequences',
headers=w2_headers,
json={'name': 'Cross Workspace', 'sequence_type': 'allele'},
)
assert w2.status_code == 200


def test_change_circularity_rejects_template_sequence(template_sequences_client):
"""Endpoints guarded by require_real_sequence return 404 for template sequences."""
c = template_sequences_client['client']
Expand Down
6 changes: 3 additions & 3 deletions uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading