Helpers for using the python-oracledb
library with the FastAPI framework.
- Install a recent Python 3.x version (if you don't already have one).
- Create a project that uses
python-oracledb
and FastAPI (if you don't already have one). - Install
fastapi-oracle
as a dependency using Poetry, pip, or similar:poetry add fastapi-oracle
- 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
- 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
To clone the repo:
git clone git@github.com: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
To build the library:
poetry build
Built by Seertech.