Skip to content
Merged
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
44 changes: 31 additions & 13 deletions app/sample_registry/src/sample_registry/registrar.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,19 +16,37 @@ class SampleRegistry(object):
machines = MACHINE_TYPES.values()
kits = ["Nextera XT"]

def __init__(self, session: Session = None):
if session:
self.session = session
else:
from sample_registry import session as imported_session

self.session = imported_session

def check_run_accession(self, acc: int) -> Run:
run = self.session.scalar(select(Run).where(Run.run_accession == acc))
if not run:
raise ValueError("Run does not exist %s" % acc)
return run
def __init__(self, session: Session = None):
if session:
self.session = session
else:
from sample_registry import session as imported_session

self.session = imported_session

def check_run_accession(self, acc: int) -> Run:
run = self.session.scalar(select(Run).where(Run.run_accession == acc))
if not run:
raise ValueError("Run does not exist %s" % acc)
return run

def get_run(self, run_accession: int) -> Run | None:
"""Return the ``Run`` record for ``run_accession``.

Parameters
----------
run_accession:
Accession number identifying the sequencing run.

Returns
-------
Run | None
The ``Run`` instance if found, otherwise ``None``.
"""

return self.session.scalar(
select(Run).where(Run.run_accession == run_accession)
)

def register_run(
self,
Expand Down
18 changes: 15 additions & 3 deletions app/sample_registry/test/test_registrar.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,9 +48,21 @@ def db() -> Generator[Session, None, None]:
session.close()


def test_check_run_accession(db):
registry = SampleRegistry(db)
assert registry.check_run_accession(1).run_accession == 1
def test_check_run_accession(db):
registry = SampleRegistry(db)
assert registry.check_run_accession(1).run_accession == 1


def test_get_run(db):
registry = SampleRegistry(db)
run = registry.get_run(1)
assert run.run_accession == 1
assert run.run_date == "2024-07-02"


def test_get_run_doesnt_exist(db):
registry = SampleRegistry(db)
assert registry.get_run(9999) is None


def test_check_run_accession_doesnt_exist(db):
Expand Down
Loading