Anysqlite
Anysqlite provides an async/await interface to the standard `sqlite3` library, supporting both Trio and Asyncio backends using the `Anyio` library. It is currently at version 0.0.5 and has an irregular or slow release cadence, with the last update in October 2023.
Warnings
- gotcha The project is at a low version (0.0.5) and has not seen recent significant development or releases (last release Oct 2023, last commit 3 years ago). This may imply a slower response to issues, potential for breaking changes in future minor releases, or unaddressed bugs.
- gotcha Anysqlite relies on `anyio` for its asynchronous backend. Ensure `anyio` is correctly installed and compatible with your chosen event loop (asyncio or trio) to prevent runtime errors related to asynchronous execution.
- gotcha As with any `sqlite3` wrapper, be mindful of potential blocking operations if complex or long-running queries are executed, especially without proper offloading. While `anysqlite` provides an async interface, the underlying `sqlite3` can still block the event loop if not used carefully.
Install
-
pip install anysqlite
Imports
- anysqlite
import anysqlite
Quickstart
import asyncio
import anysqlite
async def main():
# Connect to an in-memory database for quick testing
# Or use a file path like "my_database.db"
conn = await anysqlite.connect(":memory:")
try:
# Execute a simple query
cursor = await conn.execute("SELECT DATETIME('now')")
response = await cursor.fetchone()
print(f"Current datetime from SQLite: {response[0]}")
# Create a table and insert data
await conn.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)")
await conn.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
await conn.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))
await conn.commit()
# Fetch data
cursor = await conn.execute("SELECT id, name FROM users")
users = await cursor.fetchall()
print("Users in database:")
for user_id, user_name in users:
print(f" ID: {user_id}, Name: {user_name}")
finally:
# Ensure the connection is closed
await conn.close()
if __name__ == "__main__":
asyncio.run(main())