Today I’m presenting the database configuration that I (currently) use on pythondocs.xyz – real time interactive search of Python documentation.
It copies a SQLite database from disk into memory, so it’s very fast. It’s great for read-only workflows – dashboards and the like. It’s not suitable for sites that accept user input, as it makes no attempt to preserve updates to the database.
The config works well for pythondocs.xyz: I generate the site’s database “offline”, with a standalone parser application, and I ship the resulting database file with the web application. When the web app starts up, the database is copied into memory, and you get nice fast database access (even if your queries aren’t super efficient!)
The main dependencies are sqlalchemy, the predominant Python ORM, and aiosqlite, an async replacement for the Standard Library’s sqlite3. I use the database with FastAPI but it should work in other applications.
The database copying is handled by sqlite3‘s backup method. But sqlite3 is a synchronous library, and I want concurrent database access for performance reasons. Luckily, it’s possible to populate the database with sqlite3 and read it from aiosqlite by pointing the two libraries at the same shared memory location.
Without further ado, here’s the code that sets up the database:
from typing import Optional from sqlalchemy.engine import Engine, create_engine from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker SQLITE_SYNC_URL_PREFIX = "sqlite:///" SQLITE_ASYNC_URL_PREFIX = "sqlite+aiosqlite:///" MEMORY_LOCATION_START = "file:" MEMORY_LOCATION_END = "?mode=memory&cache=shared&uri=true" class InMemoryDatabase: """ Async in-memory SQLite DB """ def __init__(self, sql_echo: bool = False): self.sql_echo = sql_echo self._sync_memory_engine: Optional[Engine] = None self._async_memory_engine: Optional[AsyncEngine] = None self._async_sessionmaker: Optional[sessionmaker] = None def setup(self, filename: str): """ Copy DB data from disk to memory and setup async session """ sync_disk_engine = create_engine( url=SQLITE_SYNC_URL_PREFIX + filename, echo=self.sql_echo ) in_memory_url = MEMORY_LOCATION_START + filename + MEMORY_LOCATION_END # Reference to sync in-memory engine remains open self._sync_memory_engine = create_engine( url=SQLITE_SYNC_URL_PREFIX + in_memory_url, echo=self.sql_echo ) # Use sync engines to copy DB to memory backup_db(source_db=sync_disk_engine, target_db=self._sync_memory_engine) sync_disk_engine.dispose() # Create async engine at same memory location self._async_memory_engine = create_async_engine( url=SQLITE_ASYNC_URL_PREFIX + in_memory_url, echo=self.sql_echo ) self._async_sessionmaker = sessionmaker( self._async_memory_engine, class_=AsyncSession )
Compatibility with FastAPI’s dependency injection is provided by this method:
async def __call__(self) -> AsyncIterator[AsyncSession]: """Used by FastAPI Depends""" assert self._async_sessionmaker, "No sessionmaker. Run setup() first." async with self._async_sessionmaker() as session: yield session
(Thank you to the FastAPI Pagination project for inspiration!)
Use with FastAPI looks like this:
from fastapi import Depends, FastAPI from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from async_in_memory_db import InMemoryDatabase from example_data import DB_FILENAME, User app = FastAPI() db = InMemoryDatabase() @app.on_event("startup") async def setup_db(): db.setup(DB_FILENAME) @app.get("/") async def example_route(session: AsyncSession = Depends(db)) -> list[User]: results = await session.execute(select(User)) return results.scalars().all()
And here’s what you get in your web browser:
Please see the python_async_in_memory_db GitHub repo for the full code, including an example standalone query that doesn’t use FastAPI.
Is this technique useful to you? Can you see any potential pitfalls that I’ve overlooked?
Let me know in the comments below!