fastapi-oracle

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

Getting started

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

Developing

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

Building

To build the library:

poetry build

Built by Seertech.