Helpers for using the python-oracledb library with the FastAPI framework.
Project description
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.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
No source distribution files available for this release.See tutorial on generating distribution archives.
Built Distribution
Close
Hashes for fastapi_oracle-0.20.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 77b68218dbcb2ea5214463e8109bab41ab58d7481862479ee5f866eeae3066c9 |
|
MD5 | d3d1d0d2b0129c7a2726487996703ba7 |
|
BLAKE2b-256 | 6478a6d1a46017a6bfcb1a9c73ee5b640a49a18ff080f6f5ecde7c95940fde82 |