wiele APIRouterów, a SQLite

[Obsttube]

W zadaniach z pracy domowej nie ma takiej potrzeby, ale zastanawiam się co zrobić, jeśli miałbym bardziej rozbudowaną aplikację z wieloma routerami i w wielu z nich łączył się z tą samą bazą.

Teraz robię tak, że mam w routerze:

@router.on_event("startup")
async def startup():
    router.db_connection = await aiosqlite.connect('chinook.db')

Jednak w przypadku wielu routerów nie ma sensu otwierania kilku połączeń.

Da się jakoś zrobić, aby otwierać połączenie tylko w main (wtedy np. app.db_connection), a następnie mieć do tego db_connection dostęp ze wszystkich routerów?

[DziurewiczPiotr]

Hejka @Obsttube

Przyjacielu, nie wiem czy moje rozwiązanie będzie zgodne ze sztuką i czy będzie satysfakcjonujące ale tak na szybko przeklinałem coś takiego:

tutaj dla potomnych do poczytania bo zakładam że już tam byłeś: https://fastapi.tiangolo.com/tutorial/bigger-applications/#apirouter

zakładając strukturę:

.
── chinook.db
── main.py
── routers
│       ├── __init__.py
│       ├── employees.py
│       └── tracks.py

oraz employees.py

import sqlite3

from fastapi import APIRouter

from routers import connection

router = APIRouter()

@router.get("/employees")
async def employees():
    connection().row_factory = sqlite3.Row
    cursor = await connection().execute("SELECT Email FROM employees")
    data = await cursor.fetchall()
    return data

oraz tracks.py

import sqlite3

from fastapi import APIRouter

from routers import connection

router = APIRouter()

@router.get("/tracks")
async def tracks():
    connection().row_factory = sqlite3.Row
    cursor = await connection().execute("SELECT Name FROM tracks")
    data = await cursor.fetchall()
    return data

oraz main.py

import sqlite3

import aiosqlite
from fastapi import FastAPI

from routers import employees, tracks

api = FastAPI()

@api.on_event("startup")
async def startup():
    api.db_connection = await aiosqlite.connect("chinook.db")

@api.on_event("shutdown")
async def shutdown():
    await api.db_connection.close()

@api.get("/customers")
async def customers():
    api.db_connection.row_factory = sqlite3.Row
    cursor = await api.db_connection.execute("SELECT Email FROM customers")
    data = await cursor.fetchall()
    return data

api.include_router(employees.router)
api.include_router(tracks.router)

dodając do init.py

import main

def connection():
    return main.api.db_connection

po sprawdzeniu lsof'em mam:

COMMAND     PID  USER   FD   TYPE DEVICE SIZE/OFF    NODE NAME
code      69016 piotr  cwd    DIR  253,1     4096 2361954 .
bash      77439 piotr  cwd    DIR  253,1     4096 2361954 .
bash      87227 piotr  cwd    DIR  253,1     4096 2361954 .
uvicorn   87936 piotr  cwd    DIR  253,1     4096 2361954 .
python3.7 87937 piotr  cwd    DIR  253,1     4096 2361954 .
python3.7 87938 piotr  cwd    DIR  253,1     4096 2361954 .
python3.7 87938 piotr   15u   REG  253,1   884736 2361938 ./chinook.db
lsof      87976 piotr  cwd    DIR  253,1     4096 2361954 .
lsof      87977 piotr  cwd    DIR  253,1     4096 2361954 .

Nie wiem czy to jest całkowiecie legitne rozwiąznie i czy czegoś dowodzi ale strzelam, że mamy jedno połączenie do bazy.

[Obsttube]

Wielkie dzięki za odpowiedź i poświęcenie czasu na napisanie tego, właśnie czegoś takiego potrzebowałem! Jutro to zaimplementuję i przetestuję.

Przy okazji, zamiast importować aiosqlite oraz sqlite3 jednocześnie, wystarczy samo aiosqlite. Widzę, że sqlite3 użyłeś tylko do sqlite3.Row. Zamiast tego można po prostu aiosqlite.Row. (AFAIK wychodzi na to samo)

[mishioo]

Zdaje się, że łatwo można wpaść w pułapkę zapętlonych importów przy podejściu zaproponowanym przez @DziurewiczPiotr, jeśli nie jest się bardzo uważnym. Myślę, że wygodniej wydzielić funkcjonalność związaną z dostępem do bazy do osobnego modułu, niezależnego od reszty aplikacji i importować stamtąd. Podobne podejście proponuje dokumentacja FastAPI: https://fastapi.tiangolo.com/tutorial/sql-databases/ (tylko używa SQLAlchemy). Poniżej moja modyfikacja propozycji podrzuconej przez Piotra.

Struktura repo:

myapp
├─ chinook.db
├─ database.py
├─ main.py
└─ routers
         └── tracks.py

database.py

import aiosqlite

SQL_DATABASE_ADDRESS = "myapp/chinook.db"
# database connection is set up by startup event
DATABASE_CONNECTION: aiosqlite.Connection = None

async def get_db_conn():
    return DATABASE_CONNECTION

main.py

import aiosqlite
from fastapi import FastAPI, Depends
from . import database as db
from .routers import tracks

api = FastAPI()
api.include_router(tracks.router)

@api.on_event("startup")
async def startup():
    db.DATABASE_CONNECTION = await aiosqlite.connect(db.SQL_DATABASE_ADDRESS)

@api.on_event("shutdown")
async def shutdown():
    await db.DATABASE_CONNECTION.close()

@api.get("/customers")
async def customers(db_connection: aiosqlite.Connection = Depends(db.get_db_conn)):
    db_connection.row_factory = aiosqlite.Row
    cursor = await db_connection.execute("SELECT Email FROM customers")
    data = await cursor.fetchall()
    return data

routers/tracks.py

import aiosqlite
from fastapi import APIRouter, Depends
from ..database import get_db_conn

router = APIRouter()

@router.get("/tracks")
async def tracks(connection: aiosqlite.Connection = Depends(get_db_conn)):
    connection.row_factory = aiosqlite.Row
    cursor = await connection.execute("SELECT Name FROM tracks")
    data = await cursor.fetchall()
    return data

Bardzo jestem ciekaw jakie rozwiązanie stosuje się w praktyce!

[DziurewiczPiotr]

@Mishioo twoje z wyciągnięciem do database.py zdecydowanie lepsiejsze.