Skip to content
AITroveRead. Build. Understand.
Reading list
Make this comfortable

Flask SQLite application: own the request connection and verify a clean reopen

Last updated: 30 Sept 20264 min read
tutorial
IntermediateBy AITrove Editorial

A Flask request-scoped database connection can live in the application context and close when that context ends.

Download Python source kit

Operation contract

The local receipt application stores accepted records in an owned temporary SQLite file. Each request lazily opens one connection through g, uses parameterized statements and closes that connection in teardown. A duplicate receipt returns a conflict after its failed transaction rolls back. A second app instance reads the same file after the first request finishes, verifying clean persistence rather than only an in-memory response.

Failure and ownership boundary

This application has no identity or record authorization and must not be exposed as a public receipt service. JSON parsing can also collapse duplicate object keys before schema checks. Proxy body budgets, competing writers and power-loss recovery require separate tests. Flask JSON API: reject unknown fields, booleans and oversized bodies and Python SQLite project: transactional batches, duplicate IDs and reopen checks provide the preceding boundaries.

Tested environment

Dependency check: this program was executed on CPython 3.14.6 with Flask==3.1.3. Install these versions in a separate virtual environment. The download includes the recorded environment snapshot; no third-party package is part of the website runtime.

Working program

python
from pathlib import Path
import sqlite3
import tempfile
from flask import Flask, g, jsonify, request

def receipt_app(database_path):
    application = Flask(__name__)
    application.config["MAX_CONTENT_LENGTH"] = 1024
    with sqlite3.connect(database_path) as setup:
        setup.execute("CREATE TABLE IF NOT EXISTS receipts (receipt_id TEXT PRIMARY KEY, amount INTEGER NOT NULL CHECK(amount >= 0))")
    setup.close()
    def database():
        if "receipt_database" not in g:
            g.receipt_database = sqlite3.connect(database_path)
        return g.receipt_database
    @application.teardown_appcontext
    def release_database(failure):
        owned = g.pop("receipt_database", None)
        if owned is not None:
            owned.close()
    @application.post("/receipts")
    def store_receipt():
        if not request.is_json:
            return jsonify(error="JSON required"), 415
        import re
        payload = request.get_json()
        if not isinstance(payload, dict) or set(payload) != {"receipt_id", "amount"}:
            return jsonify(error="fields rejected"), 400
        if not isinstance(payload["receipt_id"], str) or re.fullmatch(r"R-[0-9]{4}", payload["receipt_id"]) is None:
            return jsonify(error="identifier rejected"), 400
        if type(payload["amount"]) is not int or not 0 <= payload["amount"] <= 1000000:
            return jsonify(error="amount rejected"), 400
        try:
            with database():
                database().execute("INSERT INTO receipts VALUES (?, ?)", (payload["receipt_id"], payload["amount"]))
        except sqlite3.IntegrityError:
            return jsonify(error="duplicate receipt"), 409
        return jsonify(receipt_id=payload["receipt_id"]), 201
    @application.get("/receipts")
    def list_receipts():
        rows = database().execute("SELECT receipt_id,amount FROM receipts ORDER BY receipt_id LIMIT 20").fetchall()
        return jsonify([{"receipt_id": identifier, "amount": amount} for identifier, amount in rows])
    return application

with tempfile.TemporaryDirectory() as directory:
    path = Path(directory) / "receipts.sqlite3"
    client = receipt_app(path).test_client()
    print("accepted:", client.post("/receipts", json={"receipt_id": "R-0041", "amount": 125}).status_code)
    print("duplicate:", client.post("/receipts", json={"receipt_id": "R-0041", "amount": 250}).status_code)
    reopened = receipt_app(path).test_client()
    print(reopened.get("/receipts").get_json())

Output

Output
accepted: 201
duplicate: 409
[{'amount': 125, 'receipt_id': 'R-0041'}]

Costs and limits

The primary-key index supports keyed uniqueness checks with backend-dependent work. The listing caps returned rows but is not complete pagination. The request lifetime bounds connection ownership; contention, disk I/O and crash durability remain deployment-dependent.

Common Mistakes

  • A connection transaction context does not close that connection automatically.
  • Persisted receipts still require identity and authorization before a public deployment.

Connected lessons

Flask JSON API: reject unknown fields, booleans and oversized bodies, Python SQLite project: transactional batches, duplicate IDs and reopen checks, Django REST Framework object permissions: check the retrieved record, not only login.

python
flask-sqlite-persistence
Storage details