fastapi-oracle 0.21.0

Helpers for using the python-oracledb library with the FastAPI framework.

fastapi-oracle

Helpers for using the python-oracledb library with the FastAPI framework.

Getting started

1. Install a recent Python 3.x version (if you don't already have one).
2. Create a project that uses python-oracledb and FastAPI (if you don't already have one).
3. Install fastapi-oracle as a dependency using Poetry, pip, or similar:

   poetry add fastapi-oracle
   

4. Set DB config in environment variables:

   DB_HOST=foodbhost.aintitfootastic.com
   DB_PORT=1521
   DB_USER=foouser
   DB_PASSWORD=foopass
   DB_SERVICE_NAME=foodb1.aintitfootastic.com
   

5. Use the utils:

   from collections.abc import AsyncIterable, Mapping
   from contextlib import asynccontextmanager
   from typing import Any, AsyncGenerator
   
   from fastapi import APIRouter, Depends, FastAPI, Request, status
   from fastapi.responses import JSONResponse
   from fastapi_oracle import (
       DbPoolConnAndCursor,
       IntermittentDatabaseError,
       close_db_pools,
       cursor_rows_as_dicts,
       cursor_rows_as_gen,
       get_db_cursor,
       handle_db_errors,
       result_keys_to_lower,
   )
   from loguru import logger
   from pydantic import BaseModel
   
   
   router = APIRouter()
   
   
   class Foo(BaseModel):
       id: int
       name: str
   
   
   async def map_list_foos_result_to_foos(
       result: AsyncIterable[Mapping[str, Any]]
   ) -> AsyncGenerator[Foo, None]:
       """Map a list foos DB result to a list of foos."""
       async for row in result:
           yield Foo(**row)
   
   
   async def list_foos_query(
       db: DbPoolConnAndCursor
   ) -> AsyncGenerator[dict[str, Any], None]:
       """List all foos."""
       cursor = await db.conn.cursor()
       await cursor.execute("SELECT id, name FROM foo")
       cursor_rows_as_dicts(cursor)
       rows = (row async for row in cursor_rows_as_gen(cursor))
       async for row in result_keys_to_lower(rows):
           yield row
   
   
   @handle_db_errors
   async def _get_foos(db: DbPoolConnAndCursor) -> list[Foo]:
       result = list_foos_query(db)
       foos = [x async for x in map_list_foos_result_to_foos(result)]
       return foos
   
   
   @router.get("/", response_model=list[Foo])
   async def read_foos(db: DbPoolConnAndCursor = Depends(get_db_cursor)):
       foos = await _get_foos(db)
       logger.info(f"Fetched {len(foos)} foos")
       return foos
   
   
   async def intermittent_database_error_handler(
       request: Request,
       exc: IntermittentDatabaseError,
   ) -> JSONResponse:
       return JSONResponse(
           status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
           content={
               "detail": [
                   {
                       "msg": f"{exc}",
                       "type": "intermittent_database_error",
                   }
               ],
           },
       )
   
   
   def get_app() -> FastAPI:
       """Create a FastAPI app instance."""
       @asynccontextmanager
       async def lifespan(app: FastAPI):
           yield
   
           await close_db_pools()
   
       app = FastAPI(
           lifespan=lifespan,
           exception_handlers={
               IntermittentDatabaseError: intermittent_database_error_handler,
           },
       )
       app.include_router(router)
       return app
   

Developing

To clone the repo:

git clone [e-mail]:Jaza/fastapi-oracle.git

To automatically fix code style issues:

./scripts/format.sh

To verify code style and static typing:

./scripts/verify.sh

To run tests:

./scripts/test.sh

To run tests that need to connect to a real database:

./scripts/testdb.sh

Building

To build the library:

poetry build

Built by Seertech.