Merge pull request #586 from ayobi/creates_barcodes_for_existing_kits

Create Working Branch

microsetta_private_api/admin/admin_impl.py

@@ -56,7 +56,7 @@ def search_kit_id(token_info, kit_id):
         admin_repo = AdminRepo(t)
         diag = admin_repo.retrieve_diagnostics_by_kit_id(kit_id)
         if diag is None:
-            return jsonify(code=404, message="Kit ID not found"), 404
+            return jsonify(code=404, message=f'Kit ID {kit_id} not found'), 404
         return jsonify(diag), 200
 
 
@@ -263,13 +263,17 @@ def create_kits(body, token_info):
     number_of_samples = body['number_of_samples']
     kit_prefix = body.get('kit_id_prefix', None)
     project_ids = body['project_ids']
+    user_barcodes = body.get('user_barcodes', [])
 
     with Transaction() as t:
         admin_repo = AdminRepo(t)
 
         try:
-            kits = admin_repo.create_kits(number_of_kits, number_of_samples,
-                                          kit_prefix, project_ids)
+            kits = admin_repo.create_kits(number_of_kits,
+                                          number_of_samples,
+                                          kit_prefix,
+                                          user_barcodes,
+                                          project_ids)
         except KeyError:
             return jsonify(code=422, message="Unable to create kits"), 422
         else:
@@ -278,6 +282,93 @@ def create_kits(body, token_info):
     return jsonify(kits), 201
 
 
+def handle_barcodes(body, token_info):
+    validate_admin_access(token_info)
+
+    kit_ids = body['kit_ids']
+
+    if isinstance(kit_ids, str):
+        kit_ids = [kit_ids]
+
+    with Transaction() as t:
+        kit_repo = KitRepo(t)
+        for kit_id in kit_ids:
+            diag = kit_repo.get_kit_all_samples(kit_id)
+            if diag is None:
+                return jsonify(f'Kit ID {kit_id} not found'), 404
+
+    action = body.get('action')
+
+    if action == 'create':
+        if 'generate_barcode_single' in body:
+            if body['generate_barcode_single']:
+                number_of_kits = 1
+                number_of_samples = 1
+        elif 'generate_barcodes' in body:
+            number_of_kits = (body['num_kits'])
+            number_of_samples = (body['num_samples'])
+        else:
+            number_of_kits = len(body['kit_ids'])
+            number_of_samples = 1
+
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+
+            barcode = admin_repo._generate_novel_barcodes(
+                number_of_kits, number_of_samples, kit_names=kit_ids)
+
+            t.commit()
+
+        return jsonify(barcode[1]), 201
+
+    elif action == 'insert':
+        return insert_barcodes(body, token_info)
+
+    else:
+        return jsonify(code=404, message='Invalid action'), 404
+
+
+def insert_barcodes(body, token_info):
+    validate_admin_access(token_info)
+
+    kit_ids = body['kit_ids']
+    barcodes = body['barcodes']
+
+    # Ensure kit_ids is a list, even if it's a single value
+    if isinstance(kit_ids, str):
+        kit_ids = [kit_ids]
+
+    with Transaction() as t:
+        admin_repo = AdminRepo(t)
+        for barcode in barcodes:
+            diag = admin_repo.retrieve_diagnostics_by_barcode(barcode)
+            if diag is not None:
+                return jsonify(f'Barcode {barcode} already exists'), 404
+
+    # Check if the lengths match
+    if len(kit_ids) != len(barcodes):
+        return jsonify("The number of kit IDs must "
+                       "match the number of barcodes"), 400
+
+    with Transaction() as t:
+        admin_repo = AdminRepo(t)
+
+        insertion_data = []
+        for kit_id, barcode in zip(kit_ids, barcodes):
+            diag = admin_repo.retrieve_diagnostics_by_kit_id(kit_id)
+            sample_info = diag['sample_diagnostic_info'][0]
+            if sample_info['projects_info']:
+                for project in sample_info['projects_info']:
+                    project_id = str(project['project_id'])
+                    insertion_data.append((kit_id, barcode, project_id))
+
+        admin_repo._insert_barcodes_to_existing_kit(insertion_data)
+
+        t.commit()
+
+    return '', 204
+
+
 def get_account_events(account_id, token_info):
     validate_admin_access(token_info)
 

microsetta_private_api/admin/sample_summary.py

@@ -83,6 +83,7 @@ def per_sample(project, barcodes, strip_sampleid):
                 ffq_complete, ffq_taken, _ = vs_repo.get_ffq_status_by_sample(
                     sample.id
                 )
+                print("ffq complete", ffq_complete, "ffq taken", ffq_taken)
 
             summary = {
                 "sampleid": None if strip_sampleid else barcode,

microsetta_private_api/admin/tests/test_admin_repo.py

@@ -352,7 +352,6 @@ def make_tz_datetime(y, m, d):
                 self.assertGreater(len(diag['projects_info']), 0)
                 self.assertEqual(len(diag['scans_info']), 2)
                 # order matters in the returned vals, so test that
-                print(diag['scans_info'][0], first_scan)
                 self.assertEqual(diag['scans_info'][0], first_scan)
                 self.assertEqual(diag['scans_info'][1], second_scan)
                 self.assertEqual(diag['latest_scan'], second_scan)
@@ -454,7 +453,11 @@ def test_get_project_barcodes_by_id(self):
             output_id = admin_repo.create_project(input)
 
             # create some fake kits
-            created = admin_repo.create_kits(2, 3, 'foo', [output_id, ])
+            created = admin_repo.create_kits(2,
+                                             3,
+                                             'foo',
+                                             None,
+                                             [output_id, ])
 
             exp = []
             for kit in created['created']:
@@ -609,14 +612,14 @@ def test_create_kits_fail_nonexistent_project(self):
                 admin_repo.create_kits(5,
                                        3,
                                        '',
+                                       None,
                                        [10000,
                                         SurveyTemplateRepo.VIOSCREEN_ID])
 
     def test_create_kits_success_not_microsetta(self):
         with Transaction() as t:
             admin_repo = AdminRepo(t)
-            non_tmi = admin_repo.create_kits(5, 3, '',
-                                             [33])
+            non_tmi = admin_repo.create_kits(5, 3, '', None, [33])
             self.assertEqual(['created', ], list(non_tmi.keys()))
             self.assertEqual(len(non_tmi['created']), 5)
             for obj in non_tmi['created']:
@@ -643,8 +646,7 @@ def test_create_kits_success_not_microsetta(self):
     def test_create_kits_success_is_microsetta(self):
         with Transaction() as t:
             admin_repo = AdminRepo(t)
-            tmi = admin_repo.create_kits(4, 2, 'foo',
-                                         [1])
+            tmi = admin_repo.create_kits(4, 2, 'foo', None, [1])
             self.assertEqual(['created', ], list(tmi.keys()))
             self.assertEqual(len(tmi['created']), 4)
             for obj in tmi['created']:
@@ -878,7 +880,6 @@ def test_scan_with_multiple_observations(self):
                 scans = [scan['observations'] for scan in diag['scans_info']]
                 scans_observation_ids = [obs['observation_id'] for scan in
                                          scans for obs in scan]
-
                 self.assertEqual(scans_observation_ids, observation_ids)
 
     def test_scan_with_wrong_observation(self):
@@ -1487,3 +1488,108 @@ def test_update_perk_fulfillment_state(self):
                 )
                 obs = cur.fetchone()
                 self.assertFalse(obs[0])
+
+    def test_generate_novel_barcodes_admin_success(self):
+        number_of_kits = 1
+        number_of_samples = 3
+
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+
+            kit_names = admin_repo._generate_novel_kit_names(
+                number_of_kits, kit_prefix=None)
+
+            new_barcodes = admin_repo._generate_novel_barcodes(
+                number_of_kits, number_of_samples, kit_names)
+
+            self.assertEqual(len(new_barcodes[1]),
+                             number_of_kits * number_of_samples)
+            self.assertTrue(all(barcodes.startswith('X')
+                            for barcodes in new_barcodes[1]))
+
+    def test_generate_novel_barcodes_admin_failure(self):
+        number_of_kits = 0
+        number_of_samples = 3
+
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+
+            kit_names = admin_repo._generate_novel_kit_names(
+                number_of_kits, kit_prefix=None)
+
+            new_barcodes = admin_repo._generate_novel_barcodes(
+                number_of_kits, number_of_samples, kit_names)
+
+            self.assertTrue(new_barcodes[1] == [], [])
+
+    def test_insert_barcodes_admin_success(self):
+        kit_name = 'test'
+        barcode = 'X00332312'
+        project_id = '1'
+        kit_name_barcode_prj_id_tuple = [(kit_name, barcode, project_id)]
+
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+            admin_repo._insert_barcodes_to_existing_kit(
+                kit_name_barcode_prj_id_tuple
+            )
+
+            with t.cursor() as cur:
+                cur.execute(
+                    "SELECT barcode "
+                    "FROM barcodes.barcode "
+                    "WHERE kit_id = %s AND barcode = %s",
+                    (kit_name, barcode)
+                )
+                obs = cur.fetchone()
+                self.assertIsNotNone(obs,
+                                     "Expected barcode not found in database")
+
+    def test_insert_barcodes_admin_fail_nonexisting_kit(self):
+        # test that inserting barcodes to a non-existent kit fails
+        kit_name_barcode_prj_id_tuple = [['test1123', 'X00332312', '1']]
+
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+            with self.assertRaises(psycopg2.errors.ForeignKeyViolation):
+                admin_repo._insert_barcodes_to_existing_kit(
+                    kit_name_barcode_prj_id_tuple)
+
+    def test_insert_barcodes_admin_fail_dup_barcodes(self):
+        # test that inserting duplicate barcode fails
+        kit_name_barcode_prj_id_tuple = [['test', '000000001', '1']]
+
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+            with self.assertRaises(psycopg2.errors.UniqueViolation):
+                admin_repo._insert_barcodes_to_existing_kit(
+                    kit_name_barcode_prj_id_tuple)
+
+    def test_user_barcode_create_kit_success(self):
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+            admin_repo.create_kits(1,
+                                   1,
+                                   '',
+                                   [['X99887769']],
+                                   [1])
+            with t.cursor() as cur:
+                cur.execute(
+                    "SELECT barcode "
+                    "FROM barcodes.barcode "
+                    "WHERE barcode = %s",
+                    ('X99887769',)
+                )
+                obs = cur.fetchall()
+                self.assertEqual(obs[0][0], 'X99887769')
+
+    def test_user_barcode_dup_create_kit_fail(self):
+        with Transaction() as t:
+            admin_repo = AdminRepo(t)
+            user_barcode = ['000000001']
+            with self.assertRaises(psycopg2.errors.UniqueViolation):
+                admin_repo.create_kits(1,
+                                       1,
+                                       '',
+                                       [user_barcode],
+                                       [1])

microsetta_private_api/api/microsetta_private_api.yaml

@@ -2805,6 +2805,47 @@ paths:
         '422':
           $ref: '#/components/responses/422UnprocessableEntity'
 
+  '/admin/add_barcodes':
+    post:
+      operationId: microsetta_private_api.admin.admin_impl.handle_barcodes
+      tags:
+        - Admin
+      summary: Create or insert barcodes
+      description: Create barcodes for an existing kit or insert barcodes into the database
+      requestBody:
+        content:
+          application/json:
+            schema:
+              type: "object"
+              properties:
+                action:
+                  type: string
+                  enum: [create, insert]
+                  description: Specify 'create' to generate new barcodes or 'insert' to insert provided barcodes
+                barcodes:
+                  type: array
+                  items:
+                    type: string
+                  description: Barcodes to insert (required if action is 'insert')
+                kit_id:
+                  type: string
+                  description: Kit ID to associate the barcodes with (required if action is 'insert')
+              required:
+                - action
+      responses:
+        '201':
+          description: Barcodes were successfully created or inserted
+          content:
+            application/json:
+              schema:
+                type: array
+        '204':
+          description: No data was returned
+        '404':
+          description: Kit ID not found or barcode already exists
+        '500':
+          description: Duplicate barcodes found (only relevant for insert action)
+
   '/admin/events/accounts/{account_id}':
     get:
       operationId: microsetta_private_api.admin.admin_impl.get_account_events