-
Notifications
You must be signed in to change notification settings - Fork 19
/
kit_repo.py
69 lines (62 loc) · 2.94 KB
/
kit_repo.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
from microsetta_private_api.repo.base_repo import BaseRepo
from microsetta_private_api.repo.sample_repo import SampleRepo
from microsetta_private_api.model.kit import Kit
class KitRepo(BaseRepo):
def __init__(self, transaction):
super().__init__(transaction)
# Kit ID is the uuid primary key of ag_kit, as opposed to supplied_kit_id,
# the more easily typed string we put on pieces of paper inside the kit
def get_kit_all_samples_by_kit_id(self, kit_id):
sample_repo = SampleRepo(self._transaction)
with self._transaction.cursor() as cur:
cur.execute("SELECT "
"ag_kit.ag_kit_id, "
"ag_kit_barcodes.ag_kit_barcode_id "
"FROM ag_kit LEFT JOIN ag_kit_barcodes ON "
"ag_kit.ag_kit_id = ag_kit_barcodes.ag_kit_id "
"WHERE "
"ag_kit.ag_kit_id = %s",
(kit_id,))
rows = cur.fetchall()
if len(rows) == 0:
return None
else:
samples = [sample_repo._get_sample_by_id(r[1]) for r in rows]
return Kit(rows[0][0], samples)
def get_kit_all_samples(self, supplied_kit_id):
sample_repo = SampleRepo(self._transaction)
with self._transaction.cursor() as cur:
cur.execute("SELECT "
"ag_kit.ag_kit_id, "
"ag_kit_barcodes.ag_kit_barcode_id "
"FROM ag_kit LEFT JOIN ag_kit_barcodes ON "
"ag_kit.ag_kit_id = ag_kit_barcodes.ag_kit_id "
"WHERE "
"ag_kit.supplied_kit_id = %s",
(supplied_kit_id,))
rows = cur.fetchall()
if len(rows) == 0:
return None
else:
samples = [sample_repo._get_sample_by_id(r[1]) for r in rows]
return Kit(rows[0][0], samples)
def get_kit_unused_samples(self, supplied_kit_id):
sample_repo = SampleRepo(self._transaction)
# Business Logic: We now define an unclaimed sample as a sample with
# a null source_id in ag_kit_barcodes
with self._transaction.cursor() as cur:
cur.execute("SELECT "
"ag_kit.ag_kit_id, "
"ag_kit_barcodes.ag_kit_barcode_id "
"FROM ag_kit LEFT JOIN ag_kit_barcodes ON "
"ag_kit.ag_kit_id = ag_kit_barcodes.ag_kit_id "
"WHERE "
"ag_kit.supplied_kit_id = %s AND "
"ag_kit_barcodes.source_id is null",
(supplied_kit_id,))
rows = cur.fetchall()
if len(rows) == 0:
return None
else:
samples = [sample_repo._get_sample_by_id(r[1]) for r in rows]
return Kit(rows[0][0], samples)