FastAPI और SQLite में database is locked: async SQLAlchemy, aiosqlite व WAL से समाधान

FastAPI में SQLite पर concurrent writes के दौरान database is locked त्रुटि से बचें: async SQLAlchemy + aiosqlite, WAL, सेशन/कमिट रणनीति और व्यावहारिक उदाहरण.

SQLite पर FastAPI समानांतर कार्यों पर आसानी से спотыкается. एक सामान्य संकेत यह है: लगभग एक ही समय पर दो endpoints को अनुरोध भेजते हैं — एक सफल हो जाता है, जबकि दूसरा db.commit() पर sqlite3.OperationalError: database is locked के साथ गिर जाता है। जब कोई writer लॉक पकड़े बैठा होता है, तो दूसरा writer आगे नहीं बढ़ सकता। अगर आपका ऐप synchronous डेटाबेस कॉल्स से event loop को रोक देता है, तो यह त्रुटि लगातार दिखाई देती है। नीचे इसका छोटा, पुनरुत्पादन योग्य उदाहरण और इसे उत्पादन में संभालने का व्यावहारिक तरीका दिया गया है।

लॉक को ट्रिगर करने वाला न्यूनतम सेटअप

उदाहरण में दो रूट हैं। पहला एक पंक्ति जोड़ता है और commit से पहले इंतजार करता है। दूसरा flush के जरिए पहले ही लॉक पकड़ लेता है, ज्यादा देर रुकता है और फिर commit करता है। दोनों async path operations हैं, लेकिन डेटाबेस सत्र synchronous है। नाम स्पष्टता के लिए बदले गए हैं, यांत्रिकी वही है।

import asyncio
import sqlite3
import threading
import time
import uuid
from loguru import logger
from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
DB_URI = "sqlite:///./test_locked.db"
sync_engine = create_engine(DB_URI, connect_args={"check_same_thread": False})
SyncSessionMaker = sessionmaker(autocommit=False, autoflush=True, bind=sync_engine)
OrmBase = declarative_base()
app = FastAPI()
class Record(OrmBase):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
OrmBase.metadata.create_all(bind=sync_engine)
def acquire_sync_session() -> Session:
    dbh = SyncSessionMaker()
    try:
        yield dbh
    finally:
        dbh.close()
@app.post("/session_a")
async def run_a(db: Session = Depends(acquire_sync_session)):
    logger.info("A start")
    token = str(uuid.uuid4())
    row = Record(name=f"session_a{token}")
    db.add(row)
    await asyncio.sleep(0.5)
    logger.info(f"A commit {token}")
    db.commit()
    return {"status": "A committed"}
@app.post("/session_b")
async def run_b(db: Session = Depends(acquire_sync_session)):
    logger.info("B start")
    await asyncio.sleep(0.1)
    token = str(uuid.uuid4())
    row = Record(name=f"session_b{token}")
    db.add(row)
    db.flush()
    logger.info(f"B flush {token}")
    await asyncio.sleep(1)
    db.commit()
    logger.info(f"B commit {token}")
    return {"status": "B committed"}
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

लॉक दिखता क्यों है

SQLite एक समय में केवल एक writer को अनुमति देता है। इस परिदृश्य में, session B flush के जरिए लिखने का लॉक जल्दी पकड़ लेता है, फिर सोते समय उसे पकड़े रखता है, और इस बीच session A commit तक पहुंचता है और लॉक से टकरा जाता है। एक और परत यह है कि यहां डेटाबेस कॉल्स synchronous हैं। भले ही route functions async हों, db.flush() और db.commit() तब तक ब्लॉक रहते हैं जब तक वे खत्म नहीं हो जाते। FastAPI में async path functions मुख्य event loop में चलते हैं, जबकि synchronous functions को threadpool में भेज दिया जाता है। Async endpoints के अंदर synchronous डेटाबेस I/O रखना event loop को ब्लॉक करता है और टाइमिंग व कंटेंशन को और खराब कर देता है। FastAPI की आधिकारिक डॉक्यूमेंटेशन में फ्रेमवर्क और event loop में async व sync के आपसी व्यवहार पर विस्तृत व्याख्या है।

राहत: डेटाबेस लेयर को वाकई async बनाएं

SQLAlchemy के async engine के साथ aiosqlite का उपयोग करने से commit और flush event loop को नियंत्रण वापस देने लगते हैं। SQLite का single-writer नियम बना रहता है, लेकिन अनावश्यक ब्लॉकिंग हट जाती है, शेड्यूलिंग बेहतर होती है, और व्यवहार में, बताए गए लोड पैटर्न पर ऐप अपेक्षित रूप से चलता है। नीचे उसी ऐप का सीधे उपयोग करने योग्य asynchronous SQLAlchemy संस्करण है:

import asyncio
import uuid
import logging
from fastapi import FastAPI, Depends
from sqlalchemy import Column, Integer, String, text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.ext.declarative import declarative_base
log = logging.getLogger("uvicorn.error")
ASYNC_DB_URI = "sqlite+aiosqlite:///./test_locked.db"
async_engine = create_async_engine(ASYNC_DB_URI, connect_args={"check_same_thread": False})
AsyncSessionMaker = async_sessionmaker(autocommit=False, autoflush=True, bind=async_engine)
AsyncBase = declarative_base()
app = FastAPI()
class RowItem(AsyncBase):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
async def bootstrap_schema():
    async with async_engine.begin() as conn:
        await conn.run_sync(AsyncBase.metadata.drop_all)
        await conn.run_sync(AsyncBase.metadata.create_all)
async def get_async_session() -> AsyncSession:
    async with AsyncSessionMaker() as sess:
        yield sess
@app.post("/session_a")
async def route_a(db: AsyncSession = Depends(get_async_session)):
    log.info("A start")
    token = str(uuid.uuid4())
    row = RowItem(name=f"session_a{token}")
    db.add(row)
    await asyncio.sleep(0.5)
    log.info(f"A commit {token}")
    await db.commit()
    return {"status": "A committed"}
@app.post("/session_b")
async def route_b(db: AsyncSession = Depends(get_async_session)):
    log.info("B start")
    await asyncio.sleep(0.1)
    token = str(uuid.uuid4())
    row = RowItem(name=f"session_b{token}")
    db.add(row)
    await db.flush()
    log.info(f"B flush {token}")
    await asyncio.sleep(1)
    await db.commit()
    log.info(f"B commit {token}")
    return {"status": "B committed"}
if __name__ == "__main__":
    import uvicorn
    asyncio.run(bootstrap_schema())
    uvicorn.run(app, host="0.0.0.0", port=8000)

इन endpoints को साथ-साथ चलाने पर अपेक्षित 200 प्रतिक्रियाएँ मिलती हैं और “database is locked” वाली विफलता नहीं होती। कुंजी यह है कि अब commit और flush awaitable हैं और event loop के साथ सहयोग करते हैं।

WAL के साथ SQLite को concurrency के अनुकूल बनाएं

जर्नल मोड को WAL (Write-Ahead Logging) पर शिफ्ट करना पढ़ने/लिखने की concurrency विशेषताओं में और सुधार कर सकता है। स्टार्टअप पर इसे इस तरह टॉगल कर सकते हैं:

from sqlalchemy import text
async def activate_wal():
    async with async_engine.connect() as conn:
        res = await conn.execute(text("PRAGMA journal_mode=WAL;"))
        print("Journal mode:", res.scalar())
# आरंभिक चरण में asyncio.run(activate_wal()) चलाएँ

SQLite के दस्तावेज़ के अनुसार, सर्वर-जैसे एक्सेस पैटर्न के लिए WAL अक्सर बेहतर होता है क्योंकि यह तेज़ है और अधिक concurrency देता है। यह single-writer नियम नहीं बदलता, लेकिन इस प्रोफ़ाइल के लिए यह एक उपयोगी ट्यूनिंग सेटिंग है।

FastAPI और SQLite के संदर्भ में यह क्यों महत्वपूर्ण है

एक async वेब स्टैक में, async endpoints के भीतर synchronous डेटाबेस कॉल्स मिलाने से ब्लॉकिंग, नाज़ुक टाइमिंग और अप्रत्याशित लॉक कंटेंशन पैदा होती है। SQLAlchemy के async engine के साथ aiosqlite का उपयोग करने से डेटाबेस ऑपरेशंस event loop के अनुरूप हो जाते हैं, अनावश्यक ब्लॉकिंग हटती है और हल्की से मध्यम concurrency में टकराते commits की संभावना घटती है। WAL सक्षम करने और autoflush=True के साथ, कम concurrency वाले परिदृश्य इस दृष्टिकोण के लिए अच्छे रहते हैं। भारी और पुख्ता concurrency की ज़रूरतों के लिए, Postgres जैसी अलग डेटाबेस तकनीक पर जाना एक तार्किक दिशा है। व्यावहारिक तरीके से क्षमता को आँकने के लिए, ऐप चालू करें, लोड टेस्ट चलाएँ और देखें कि बड़ी संख्या में अनुरोध उतनी ही संख्या में रिकॉर्ड्स के रूप में दर्ज हो रहे हैं या नहीं।

निष्कर्ष

अगर FastAPI के साथ SQLite में concurrent writes पर “database is locked” दिखे, तो स्थिरता तक पहुँचने का सबसे तेज़ तरीका है डेटा लेयर को aiosqlite के साथ asynchronous बनाना और WAL सक्षम करना। FastAPI के async मॉडल को ध्यान में रखें: async path functions मुख्य event loop में चलते हैं, जबकि sync कोड threadpool में जाता है। जब सब कुछ async रहता है, commit और flush yield कर सकते हैं और scheduler अक्सर सही निर्णय लेता है। फिर भी लॉक सीमा से टकराएँ, तो SQLite सेटिंग्स जैसे WAL और connection timeout पर विचार करें और महत्वपूर्ण writes के आसपास साधारण try/except लागू करें। यदि आपका लोड कम concurrency से आगे बढ़ जाए, तो multi-writer परिदृश्यों के लिए डिज़ाइन किए गए डेटाबेस का मूल्यांकन करें।