中文
devexamples

SQLite Connection With Context Manager

Open a SQLite connection as a context manager so transactions commit on success and roll back on error, and the connection always closes.

PythonData Fetching

Code

snippet
import sqlite3
from contextlib import closing
from typing import List, Tuple


def fetch_users(db_path: str) -> List[Tuple[int, str]]:
    with closing(sqlite3.connect(db_path)) as connection:
        connection.row_factory = sqlite3.Row
        with connection:
            cursor = connection.execute(
                "SELECT id, email FROM users WHERE active = ?", (1,)
            )
            rows = cursor.fetchall()
    return [(row["id"], row["email"]) for row in rows]


def create_schema(connection: sqlite3.Connection) -> None:
    connection.execute(
        "CREATE TABLE IF NOT EXISTS users ("
        "id INTEGER PRIMARY KEY, email TEXT NOT NULL, active INTEGER DEFAULT 0)"
    )

Related Examples