fastapi-oracle 0.25.0
Helpers for using the python-oracledb library with the FastAPI framework.
License: Apache-2.0
Requires Python: <4,>=3.10

# fastapi-oracle

Helpers for using the [`python-oracledb`](https://github.com/oracle/python-oracledb) library with the [FastAPI](https://github.com/tiangolo/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](https://python-poetry.org/), pip, or similar:
   ```sh
   poetry add fastapi-oracle
   ```
4. Set DB config in environment variables:
   ```sh
   DB_HOST=foodbhost.aintitfootastic.com
   DB_PORT=1521
   DB_USER=foouser
   DB_PASSWORD=foopass
   DB_SERVICE_NAME=foodb1.aintitfootastic.com
5. Use the utils:
   ```python
   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:

```sh
git clone git@github.com:Jaza/fastapi-oracle.git
```

To automatically fix code style issues:

```sh
./scripts/format.sh
```

To verify code style and static typing:

```sh
./scripts/verify.sh
```

To run tests:

```sh
./scripts/test.sh
```

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

```sh
./scripts/testdb.sh
```


## Building

To build the library:

```sh
poetry build
```


Built by [Seertech](https://www.seertechsolutions.com/).