from collections.abc import AsyncIterator from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.exceptions import HTTPException from fastapi.middleware.cors import CORSMiddleware # from fastapi.requests import Request from fastapi.responses import JSONResponse from fastapi_cache import FastAPICache from fastapi_cache.backends.redis import RedisBackend from fastapi_cache.decorator import cache from logging_config import logger from redis import asyncio as aioredis from request_response_models import BasicError, ConvertRequest, MolFileModel @asynccontextmanager async def lifespan(_: FastAPI) -> AsyncIterator[None]: redis = aioredis.from_url("redis://redis:6379") FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache") yield app = FastAPI(lifespan=lifespan) origins = [ "http://localhost", "http://localhost:8001", ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.post( "/convert", responses={ 200: {"description": "Coversion Successful", "model": MolFileModel}, 400: {"description": "Item created", "model": BasicError}, }, ) def convert_molecule(req: ConvertRequest): try: # Rad the string and format from request mol = pybel.readstring(req.format, req.text) if req.add_hydrogen: mol.addh() if req.convert_3d: mol.make3D() if req.optimize_geometry: mol.localopt() logger.info(f"Converting from format {req.format}") # To compute the Number of electrons and orbitals atoms = [(atom.atomicnum, atom.coords) for atom in mol.atoms] mol_pyscf = gto.Mole() mol_pyscf.atom = atoms mol_pyscf.basis = "sto-3g" # simple basis mol_pyscf.charge = mol.charge # default net charge mol_pyscf.spin = mol.spin - 1 # multiplicity - 1 mol_pyscf.build() # Export the resulting molecule mol.OBMol.SetTitle( f"Charge={mol.charge} Multiplicity={mol.spin} Electrons={mol_pyscf.nelectron} Orbitals={mol_pyscf.nao_nr()}" ) mol2string = mol.write("xyz") return JSONResponse({"molfile": mol2string}, 200) except Exception as e: raise HTTPException(status_code=400, detail={"error": str(e)}) @app.get("/informats") @cache() async def get_informats(): return pybel.informats @app.get( "/health", responses={ 200: { "description": "Is the service running?", "content": {"application/json": {"example": {"status": "healthy"}}}, }, }, ) @cache() async def health_check(): return {"status": "healthy"}