Recommended way to use FastAPI with SQL Server 2019

[jdelabr]

Hello,

we've been having some really good fun using FastAPI, so congratulations to the community on delivering such a marvel.
Issue is we use pyodbc DBAPI to connecto to MS SQL Server 2019 and that's not possible to make any async calls from what I understand. So we end up seeing our users queued and sometimes even get timeouts.

Do you know if there are ways to use SQL server in FastAPI? The thread below does not seem to mention it.
https://fastapi.tiangolo.com/advanced/async-sql-databases/
I cannot really find any clue on what DBAPI to use so fearing that may not be supported.

Thank you in advance for the help!

Julien

[Mause]

Have you tried https://github.com/aio-libs/aioodbc ?

[jdelabr]

Than you Mause,

this one worked like a charm!

Julien

[tiangolo]

Thanks for the help here @Mause ! 👏 🙇

Thanks for reporting back and closing the issue @ravyouli 👍

[basiltt]

Hi Ravyouli, Can I see how did you manage to integrate aioodbc with FastAPI ?

[jordantshaw]

@ravyouli Are you implementing aiodbc with sqlalchemy? Curious how you implemented.

[sahil0094]

@ravyouli can you pls share snippet of how you used aiodbc in fastapi

[NILodio]

@ravyouli can you please share information about this implementation!

[basiltt]

You can refer to below code, if it helps.

from typing import List

import aioodbc
from aioodbc.cursor import Cursor
from fastapi import HTTPException
from pyodbc import OperationalError
from concurrent.futures import ThreadPoolExecutor
import asyncio
from starlette import status
from fastapi import APIRouter, Depends, Query


loop = asyncio.get_event_loop()

router = APIRouter(
    prefix='/payroll_reporting',
    tags=['PayrollReporting'])


async def get_cursor() -> Cursor:
    dsn = rf'Driver={DB_DRIVER};Server={DB_SERVER},{DB_PORT};Database={DB_DATABASE};UID={DB_USERNAME};PWD={DB_PASSWORD};Trusted_Connection=no;'
    try:
        async with aioodbc.connect(dsn=dsn, loop=loop, executor=ThreadPoolExecutor(max_workers=50)) as conn:
            async with conn.cursor() as cur:
                yield cur
    except OperationalError:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY,
                            detail="DB connectivity failed")


def get_quoted(value):
    return f"'{value}'"


async def get_fallout_details_db(db: Cursor, country: List,
                                 start_interface_load_date: str,
                                 end_interface_load_date: str,
                                 record_status=False):
    if record_status:
        query = fr"select * FROM [PRISM].[PR_fallout_output] where Country in ({(', '.join(map(lambda x: get_quoted(x), country)))}) and Interface__Load_Date >= '{start_interface_load_date}' and Interface__Load_Date <= '{end_interface_load_date}' and Record_Status = 'Open';"
    else:
        query = fr"select * FROM [PRISM].[PR_fallout_output] where Country in ({(', '.join(map(lambda x: get_quoted(x), country)))}) and Interface__Load_Date >= '{start_interface_load_date}' and Interface__Load_Date <= '{end_interface_load_date}';"
    await db.execute(query)
    records = await db.fetchall()
    column_names = [column[0] for column in db.description]
    out = []
    for record in records:
        out.append(dict(zip(column_names, record)))
    return out


@router.get('/get_fallout_details', status_code=status.HTTP_200_OK)
async def get_fallout_details(start_interface_load_date: str,
                              end_interface_load_date: str,
                              record_status=False, db: Cursor = Depends(get_cursor),
                              country: List[str] = Query(...)):
    return await get_fallout_details_db(country=country, start_interface_load_date=start_interface_load_date,
                                        end_interface_load_date=end_interface_load_date, record_status=record_status,
                                        db=db)

[obronco]

I found it easier to use the lifespan parameter of the FastAPI app:

In main.py:
from database import db_pool_lifespan

app = FastAPI(lifespan=db_pool_lifespan)

My database.py:

from aioodbc import create_pool
from contextlib import asynccontextmanager
from dotenv import dotenv_values
from fastapi import FastAPI

config = dotenv_values('.env')
pool = None

@asynccontextmanager
async def db_pool_lifespan(app: FastAPI):
global pool
_dsn =
"Driver=ODBC Driver 18 for SQL Server;"
f"Server={config['db_host']},1433;"
f"Database={config['db_use']};"
f"UID={config['db_usr']};"
f"PWD={config['db_pwd']};"
"Trusted_Connection=no;"
pool = await create_pool(dsn=_dsn)
yield
pool.close()
await pool.wait_closed()

async def select_query(stm, cstr=False):
results = []
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute(stm)
cols = [c[0] for c in cur.description]
rows = await cur.fetchall()
for row in rows:
results.append({
col : str(row[i]) if cstr else row[i] for i, col in enumerate(cols)
})
return results

async def execute_query(stm: str, params: tuple):
row_count = 0
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute(stm, params)
await cur.execute('select @@rowcount')
row_count = await cur.fetchall()
return row_count[0][0]

[github-actions[bot]]

Assuming the original need was handled, this will be automatically closed now. But feel free to add more comments or create new issues or PRs.