sij/sijapi
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Auto-update: Thu Jul 25 01:32:50 PDT 2024

Browse source
This commit is contained in:
sanj 2024-07-25 01:32:50 -07:00
parent 5d6428302f
commit 98ec9561aa
1 changed files with 5 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
sijapi/classes.py
View file

@ -31,9 +31,9 @@ def warn(text: str): logger.warning(text)
def err(text: str): logger.error(text) def err(text: str): logger.error(text)
def crit(text: str): logger.critical(text) def crit(text: str): logger.critical(text)
TS_ID=os.getenv("TS_ID", "NULL")
T = TypeVar('T', bound='Configuration') T = TypeVar('T', bound='Configuration')
TS_ID = os.environ.get('TS_ID')
class Configuration(BaseModel): class Configuration(BaseModel):
HOME: Path = Path.home() HOME: Path = Path.home()
_dir_config: Optional['Configuration'] = None _dir_config: Optional['Configuration'] = None
@ -280,7 +280,10 @@ class APIConfig(BaseModel):
@property @property
def local_db(self): def local_db(self):
return next((db for db in self.POOL if db['ts_id'] == TS_ID), None) local_db = next((db for db in self.POOL if db['ts_id'] == TS_ID), None)
if local_db is None:
raise ValueError(f"No database configuration found for TS_ID: {TS_ID}")
return local_db
@asynccontextmanager @asynccontextmanager
async def get_connection(self, pool_entry: Dict[str, Any] = None): async def get_connection(self, pool_entry: Dict[str, Any] = None):