tests for tars

Author: mekhlakapoor

workflow_templates/tag_and_publish.yml .github/workflows/tag_and_publish.ymlworkflow_templates/tag_and_publish.yml renamed to .github/workflows/tag_and_publish.yml

@@ -1,9 +1,10 @@
 """Module for settings to connect to TARS backend"""
 
-from azure.identity import ClientSecretCredential
+from typing import Optional
+
 from pydantic import Field, SecretStr
 from pydantic_settings import BaseSettings, SettingsConfigDict
-from typing import Optional
+
 
 class Settings(BaseSettings):
     """Settings needed to connect to LabTracks Database"""
@@ -22,16 +23,6 @@ class Settings(BaseSettings):
     scope: str = Field(..., description="Scope")
     resource: str = Field(..., description="Resource")
     redis_url: Optional[str] = Field(default=None)
-    
-    def get_bearer_token(self):
-        """Retrieves Access Token"""
-        credentials = ClientSecretCredential(
-            tenant_id=self.tenant_id,
-            client_id=self.client_id,
-            client_secret=self.client_secret.get_secret_value(),
-        )
-        token, exp = credentials.get_token(self.scope)
-        return token, exp
 
 
 def get_settings() -> Settings:

workflow_templates/test_server.yml .github/workflows/test_server.ymlworkflow_templates/test_server.yml renamed to .github/workflows/test_server.yml

@@ -118,11 +118,14 @@ def get_prep_lot_data(self, prep_lot_id: str) -> List[PrepLotData]:
         response = self._get_prep_lot_response(prep_lot_id=sanitized_id)
         prep_lot_data = []
         for d in response.data:
-            if isinstance(d.lot, str) and d.lot.strip().lower() == sanitized_id.lower():
+            if (
+                isinstance(d.lot, str)
+                and d.lot.strip().lower() == sanitized_id.lower()
+            ):
                 prep_lot_data.append(d)
         return prep_lot_data
 
-    def _get_raw_molecules_response(self, plasmid_name: str) -> Response:
+    def _get_raw_molecule_response(self, plasmid_name: str) -> Response:
         """
         Get raw requests Response
         Parameters
@@ -161,7 +164,7 @@ def _get_molecule_response(self, plasmid_name: str) -> MoleculeResponse:
 
         """
 
-        response = self._get_raw_molecules_response(plasmid_name=plasmid_name)
+        response = self._get_raw_molecule_response(plasmid_name=plasmid_name)
         response.raise_for_status()
         model_response = MoleculeResponse.model_validate_json(
             json.dumps(response.json())
@@ -256,29 +259,3 @@ def get_virus_data(self, virus_name: str) -> List[Virus]:
                 else:
                     continue
         return virus_data
-    
-    def get_injection_material_data(self, prep_lot_id: str):
-        """
-        Return injection material data for a prep_lot_id
-        Parameters
-        ----------
-        prep_lot_id : str
-
-        Returns
-        -------
-        List[PrepLotData]
-        """
-        prep_lot_data = self.get_prep_lot_data(prep_lot_id=prep_lot_id)
-        for lot in prep_lot_data:
-            virus_tars_id = next(
-                    (
-                        alias["name"]
-                        for alias in lot["viralPrep"]["virus"]["aliases"]
-                        if alias["isPreferred"]
-                    ),
-                    None,
-                )
-            # check virus registry with tars id
-            virus_data = self.get_virus_data(virus_name=virus_tars_id)
-
-            

aind-tars-service-server/src/aind_tars_service_server/configs.py

@@ -2,23 +2,23 @@
 
 import logging
 import os
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.routing import APIRoute
-from collections.abc import AsyncIterator
-from contextlib import asynccontextmanager
-
-from aind_tars_service_server import __version__ as service_version
-from aind_tars_service_server.route import router
-from aind_tars_service_server.configs import get_settings
 from fastapi_cache import FastAPICache
 from fastapi_cache.backends.inmemory import InMemoryBackend
 from fastapi_cache.backends.redis import RedisBackend
 
 # from redis import asyncio as aioredis  # noqa
 from redis.asyncio import from_url  # noqa
 
+from aind_tars_service_server import __version__ as service_version
+from aind_tars_service_server.configs import get_settings
+from aind_tars_service_server.route import router
+
 # The log level can be set by adding an environment variable before launch.
 log_level = os.getenv("LOG_LEVEL", "INFO")
 logging.basicConfig(level=log_level)
@@ -30,6 +30,7 @@
 
 """
 
+
 @asynccontextmanager
 async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     """Init cache and add to lifespan of app"""
@@ -41,6 +42,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         FastAPICache.init(InMemoryBackend(), prefix="fastapi-cache")
     yield
 
+
 # noinspection PyTypeChecker
 app = FastAPI(
     title="aind-tars-service",

aind-tars-service-server/src/aind_tars_service_server/handler.py

@@ -1,17 +1,20 @@
 """Module for defining models from TARS"""
 
 from datetime import datetime
-from typing import Any, List, Optional, Literal
+from typing import Any, List, Literal, Optional
 
 from pydantic import BaseModel, Field
+
 from aind_tars_service_server import __version__
 
+
 class HealthCheck(BaseModel):
     """Response model to validate and return when performing a health check."""
 
     status: Literal["OK"] = "OK"
     service_version: str = __version__
 
+
 class BaseResponse(BaseModel):
     """Most responses have this shape"""
 
@@ -51,7 +54,7 @@ class Virus(DataModel):
     aliases: List[Alias] = Field(default=[])
     capsid: Optional[Any] = Field(default=None)
     citations: list = Field(default=None)
-    molecules: list = Field(default=None) # List[Molecule]? 
+    molecules: list = Field(default=None)  # List[Molecule]?
     otherMolecules: list = Field(default=None)
     patents: list = Field(default=None)
 
@@ -171,8 +174,9 @@ class VirusResponse(BaseResponse):
 
     data: List[Virus] = Field(default=[])
 
+
 class InjectionMaterialData(BaseModel):
     """Tars Injection Material Data"""
 
     prep_lot_data: List[PrepLotData] = Field(default=[])
-    virus_data: List[Virus] = Field(default=[])
+    virus_data: List[Virus] = Field(default=[])

aind-tars-service-server/src/aind_tars_service_server/main.py

@@ -1,17 +1,26 @@
 """Module to handle endpoint responses"""
 
+from typing import List
+
+from azure.core.credentials import AccessToken
+from azure.identity import ClientSecretCredential
 from fastapi import APIRouter, Depends, HTTPException, Path, status
-from aind_tars_service_server.handler import SessionHandler
-from aind_tars_service_server.models import HealthCheck
-from aind_tars_service_server.session import get_session
-from aind_tars_service_server.configs import Settings, get_settings
-from aind_tars_service_server.models import PrepLotData, MoleculeData, Virus
 from fastapi_cache.decorator import cache
 from requests import Session
-from typing import List
+
+from aind_tars_service_server.configs import Settings, get_settings
+from aind_tars_service_server.handler import SessionHandler
+from aind_tars_service_server.models import (
+    HealthCheck,
+    MoleculeData,
+    PrepLotData,
+    Virus,
+)
+from aind_tars_service_server.session import get_session
 
 router = APIRouter()
 
+
 @router.get(
     "/healthcheck",
     tags=["healthcheck"],
@@ -44,8 +53,12 @@ async def get_access_token(settings: Settings) -> str:
     str
 
     """
-    token, _ = settings.get_bearer_token()
-    return token
+    credentials: AccessToken = ClientSecretCredential(
+        tenant_id=settings.tenant_id,
+        client_id=settings.client_id,
+        client_secret=settings.client_secret.get_secret_value(),
+    ).get_token(settings.scope)
+    return credentials.token
 
 
 @router.get(
@@ -58,8 +71,8 @@ async def get_viral_prep_lot(
     settings: Settings = Depends(get_settings),
 ):
     """
-    ## TARS Endpoint to retrieve injection materials.
-    Retrieves injection materials information from TARS for a prep_lot_id.
+    ## TARS Endpoint to retrieve viral prep lot data.
+    Retrieves viral prep lot information from TARS for a prep_lot_id.
     """
     bearer_token = await get_access_token(settings=settings)
     handler = SessionHandler(
@@ -69,10 +82,11 @@ async def get_viral_prep_lot(
     if len(prep_lot_data) == 0:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"Prep lot with ID {prep_lot_id} not found"
+            detail=f"Prep lot with ID {prep_lot_id} not found",
         )
     return prep_lot_data
 
+
 @router.get(
     "/molecule/{plasmid_name}",
     response_model=List[MoleculeData],
@@ -94,10 +108,11 @@ async def get_molecule_data(
     if len(molecule_data) == 0:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"Molecule data for plasmid with name {plasmid_name} not found"
+            detail=f"Molecule data for {plasmid_name} not found",
         )
     return molecule_data
 
+
 @router.get(
     "/virus/{virus_name}",
     response_model=List[Virus],
@@ -119,32 +134,6 @@ async def get_virus_data(
     if len(virus_data) == 0:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
-            detail=f"Virus data for {virus_name} not found"
+            detail=f"Virus data for {virus_name} not found",
         )
     return virus_data
-
-# @router.get(
-#     "/injection_materials/{prep_lot_id}",
-#     response_model=List[PrepLotData],
-# )
-# async def get_injection_materials(
-#     prep_lot_id: str = Path(..., examples=["VT3214g"]),
-#     session: Session = Depends(get_session),
-#     settings: Settings = Depends(get_settings),
-# ):
-#     """
-#     ## TARS Endpoint to retrieve injection materials.
-#     Retrieves injection materials information from TARS for a prep_lot_id.
-#     """
-#     bearer_token = await get_access_token(settings=settings)
-#     handler = SessionHandler(
-#         session=session, bearer_token=bearer_token, settings=settings
-#     )
-    
-#     injection_materials = handler.get_injection_materials(prep_lot_id=prep_lot_id)
-#     if len(injection_materials) == 0:
-#         raise HTTPException(
-#             status_code=status.HTTP_404_NOT_FOUND,
-#             detail=f"Injection materials for prep lot with ID {prep_lot_id} not found"
-#         )
-#     return injection_materials