0%

Persistent Data: Working with SQLite in Python

Store data permanently. Query with SQL. No external database server needed. Perfect for configuration, logs, and application data.

Your Python programs work with data. Lists, dictionaries, variables. But when the program ends, the data disappears. The next run starts from scratch. How do you save data permanently? How do you search through thousands of records? How do you update one field without rewriting the entire file?
SQLite is the answer. It is a lightweight, file-based database. No server to install. No configuration. Just a file on your disk. You interact with it using SQL (Structured Query Language). SQLite is built into Python. No extra installation needed.
This tutorial teaches you to use SQLite in Python. You will learn to create databases, design tables, insert, query, update, and delete data. You will also learn to use context managers for safe connections and avoid common security pitfalls like SQL injection.

🕯️ Magic Note

SQLite is the most widely deployed database in the world. It is in every Android and iOS device, every Mac and Windows computer, every web browser (Chrome, Firefox, Safari). It is not just for small projects—it can handle terabyte-sized databases and thousands of concurrent reads.

What is SQLite?
Understanding where SQLite fits in the database world.
FeatureSQLiteTraditional Databases (PostgreSQLMySQL)
ServerNo server, file-basedRequires running server process
SetupZero configurationInstallation and configuration required
ConcurrencyLimited write concurrencyHigh write concurrency
Use caseEmbedded, mobile, desktopWeb applications, multi-user systems
PortabilitySingle file, copy anywhereRequires dump/restore
PythonBuilt-in (`import sqlite3`)Requires third-party driver

🕯️ Magic Note

SQLite stores the entire database in a single file. You can copy, email, or backup this file. The database is portable across operating systems. This simplicity makes SQLite perfect for desktop applications, mobile apps, and small to medium web applications.

Basic SQLite Operations
Connect to a database and execute SQL commands.

Python (basic_operations.py)

import sqlite3

# Connect to database (creates file if not exists)

conn = sqlite3.connect(“mydatabase.db”)

cursor = conn.cursor()

# Create a table

cursor.execute(“””

CREATE TABLE IF NOT EXISTS users (

id INTEGER PRIMARY KEY AUTOINCREMENT,

name TEXT NOT NULL,

age INTEGER,

email TEXT UNIQUE

)

“””)

# Insert data

cursor.execute(“INSERT INTO users (name, age, email) VALUES (?, ?, ?)”,

(“Ali Rezaei”, 25, “ali@example.com”))

cursor.execute(“INSERT INTO users (name, age, email) VALUES (?, ?, ?)”,

(“Sara Mohammadi”, 30, “sara@example.com”))

# Commit changes (important!)

conn.commit()

# Query data

cursor.execute(“SELECT * FROM users”)

rows = cursor.fetchall()

for row in rows:

print(row)

# Close connection

conn.close()

💡 Always use `?` placeholders in SQL queries. Never use string formatting (f-strings) to insert values. This prevents SQL injection attacks.
Context Managers for Database Connections
Use `with` statements to automatically commit and close connections.

Python (with_context_manager.py)

import sqlite3

from contextlib import contextmanager

@contextmanager

def get_db_connection(db_name=”mydatabase.db”):

conn = sqlite3.connect(db_name)

conn.row_factory = sqlite3.Row # Access columns by name

try:

yield conn

conn.commit()

except Exception as e:

conn.rollback()

raise e

finally:

conn.close()

# Usage

with get_db_connection() as conn:

cursor = conn.execute(“SELECT * FROM users WHERE age > ?”, (20,))

for row in cursor:

print(f”Name: {row[‘name’]}, Age: {row[‘age’]}”)

🕯️ Magic Note

The `row_factory = sqlite3.Row` setting allows you to access columns by name (`row[“name”]`) instead of index (`row[0]`). This makes your code much more readable.

CRUD Operations: Create, Read, Update, Delete
Complete example with all basic database operations.

Python (crud_operations.py)

import sqlite3

from contextlib import contextmanager

@contextmanager

def get_db():

conn = sqlite3.connect(“products.db”)

conn.row_factory = sqlite3.Row

try:

yield conn

conn.commit()

except:

conn.rollback()

raise

finally:

conn.close()

# Initialize table

def init_db():

with get_db() as conn:

conn.execute(“””

CREATE TABLE IF NOT EXISTS products (

id INTEGER PRIMARY KEY AUTOINCREMENT,

name TEXT NOT NULL,

price REAL NOT NULL,

stock INTEGER DEFAULT 0,

category TEXT

)

“””)

# CREATE

def create_product(name, price, stock, category):

with get_db() as conn:

cursor = conn.execute(“””

INSERT INTO products (name, price, stock, category)

VALUES (?, ?, ?, ?)

“””, (name, price, stock, category))

return cursor.lastrowid

# READ (all)

def get_all_products():

with get_db() as conn:

cursor = conn.execute(“SELECT * FROM products ORDER BY id”)

return [dict(row) for row in cursor.fetchall()]

# READ (single)

def get_product_by_id(product_id):

with get_db() as conn:

cursor = conn.execute(“SELECT * FROM products WHERE id = ?”, (product_id,))

row = cursor.fetchone()

return dict(row) if row else None

# READ (filtered)

def get_products_by_category(category):

with get_db() as conn:

cursor = conn.execute(“SELECT * FROM products WHERE category = ?”, (category,))

return [dict(row) for row in cursor.fetchall()]

# UPDATE

def update_product_stock(product_id, new_stock):

with get_db() as conn:

conn.execute(“UPDATE products SET stock = ? WHERE id = ?”, (new_stock, product_id))

return True

# DELETE

def delete_product(product_id):

with get_db() as conn:

conn.execute(“DELETE FROM products WHERE id = ?”, (product_id,))

return True

# Usage example

if __name__ == “__main__”:

init_db()

# Create

create_product(“Python Book”, 29.99, 50, “Books”)

create_product(“Coffee Mug”, 12.99, 100, “Kitchen”)

create_product(“Notebook”, 5.99, 200, “Stationery”)

# Read

products = get_all_products()

for p in products:

print(f”{p[‘name’]}: ${p[‘price’]} (stock: {p[‘stock’]})”)

# Update

update_product_stock(1, 45)

# Filtered read

books = get_products_by_category(“Books”)

print(f”\nBooks: {books}”)

Working with SQLite Row Factory
Access columns by name for cleaner code.

Python (row_factory_demo.py)

import sqlite3

conn = sqlite3.connect(“example.db”)

conn.row_factory = sqlite3.Row

cursor = conn.cursor()

cursor.execute(“CREATE TABLE IF NOT EXISTS users (id INTEGER, name TEXT, age INTEGER)”)

cursor.execute(“INSERT INTO users VALUES (1, ‘Ali’, 25)”)

cursor.execute(“INSERT INTO users VALUES (2, ‘Sara’, 30)”)

conn.commit()

cursor.execute(“SELECT * FROM users”)

for row in cursor:

# Access by index (hard to read)

print(f”Index: {row[0]}, {row[1]}, {row[2]}”)

# Access by name (clear and maintainable)

print(f”Named: {row[‘id’]}, {row[‘name’]}, {row[‘age’]}”)

# Convert to dictionary for flexibility

user_dict = dict(row)

print(f”Dict: {user_dict}”)

conn.close()

SQL Injection Prevention (Critical!)
Never use string formatting in SQL queries. Always use parameterized queries.

Python (dangerous.py – DO NOT DO THIS)

import sqlite3

conn = sqlite3.connect(“users.db”)

cursor = conn.cursor()

# DANGEROUS: SQL injection vulnerability

user_input = “Robert’; DROP TABLE users; –“

query = f”SELECT * FROM users WHERE name = ‘{user_input}'”

cursor.execute(query) # This would delete the users table!

Python (safe.py – ALWAYS DO THIS)

import sqlite3

conn = sqlite3.connect(“users.db”)

cursor = conn.cursor()

# SAFE: Parameterized query with ? placeholders

user_input = “Robert’; DROP TABLE users; –“

cursor.execute(“SELECT * FROM users WHERE name = ?”, (user_input,))

# The input is treated as a string, not executable SQL

rows = cursor.fetchall()

print(f”Found {len(rows)} users”) # No deletion occurs

⚠️ Always use `?` placeholders and pass parameters as a tuple. Never use f-strings, `.format()`, or concatenation to build SQL queries. SQL injection is one of the most common and dangerous security vulnerabilities.
Transactions and Rollback
Group multiple operations into atomic transactions.

Python (transactions.py)

import sqlite3

def transfer_money(conn, from_account, to_account, amount):

try:

# Start transaction (SQLite auto-starts with first DML)

cursor = conn.cursor()

# Withdraw

cursor.execute(“UPDATE accounts SET balance = balance – ? WHERE id = ?”, (amount, from_account))

# Deposit

cursor.execute(“UPDATE accounts SET balance = balance + ? WHERE id = ?”, (amount, to_account))

# If we reach here, commit all changes

conn.commit()

print(“Transfer successful”)

except Exception as e:

# Any error rolls back ALL changes

conn.rollback()

print(f”Transfer failed: {e}”)

# Usage

conn = sqlite3.connect(“bank.db”)

conn.execute(“CREATE TABLE IF NOT EXISTS accounts (id INTEGER, balance REAL)”)

conn.execute(“INSERT OR REPLACE INTO accounts VALUES (1, 1000)”)

conn.execute(“INSERT OR REPLACE INTO accounts VALUES (2, 500)”)

conn.commit()

transfer_money(conn, 1, 2, 200)

conn.close()

🕯️ Magic Note

SQLite automatically starts a transaction with the first INSERT, UPDATE, or DELETE. If you call `commit()`, all changes are saved. If you call `rollback()` (or an error occurs), all changes since the last commit are discarded. This ensures data consistency.

Backup and Restore
Simple backup and restore operations for SQLite databases.

Python (backup.py)

import sqlite3

import shutil

from datetime import datetime

def backup_database(db_path=”mydatabase.db”):

“””Create a timestamped backup of the database.”””

timestamp = datetime.now().strftime(“%Y%m%d_%H%M%S”)

backup_path = f”{db_path}.{timestamp}.backup”

# Method 1: File copy (simplest)

shutil.copy2(db_path, backup_path)

print(f”Backup saved to {backup_path}”)

return backup_path

def backup_via_sqlite(db_path=”mydatabase.db”):

“””Backup using SQLite’s online backup API (safer for live databases).”””

timestamp = datetime.now().strftime(“%Y%m%d_%H%M%S”)

backup_path = f”{db_path}.{timestamp}.backup”

source = sqlite3.connect(db_path)

destination = sqlite3.connect(backup_path)

source.backup(destination)

destination.close()

source.close()

print(f”Backup saved to {backup_path}”)

return backup_path

Practical Example: Task Manager Database
Complete example of a task management system using SQLite.

Python (task_manager.py)

import sqlite3

from contextlib import contextmanager

from datetime import datetime

@contextmanager

def get_db():

conn = sqlite3.connect(“tasks.db”)

conn.row_factory = sqlite3.Row

try:

yield conn

conn.commit()

except:

conn.rollback()

raise

finally:

conn.close()

def init_db():

with get_db() as conn:

conn.execute(“””

CREATE TABLE IF NOT EXISTS tasks (

id INTEGER PRIMARY KEY AUTOINCREMENT,

title TEXT NOT NULL,

description TEXT,

status TEXT DEFAULT ‘pending’,

priority INTEGER DEFAULT 1,

created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,

completed_at TIMESTAMP

)

“””)

class TaskManager:

@staticmethod

def add_task(title, description=””, priority=1):

with get_db() as conn:

cursor = conn.execute(“””

INSERT INTO tasks (title, description, priority)

VALUES (?, ?, ?)

“””, (title, description, priority))

return cursor.lastrowid

@staticmethod

def get_all_tasks():

with get_db() as conn:

cursor = conn.execute(“SELECT * FROM tasks ORDER BY priority DESC, created_at”)

return [dict(row) for row in cursor.fetchall()]

@staticmethod

def get_pending_tasks():

with get_db() as conn:

cursor = conn.execute(“SELECT * FROM tasks WHERE status = ‘pending’ ORDER BY priority DESC”)

return [dict(row) for row in cursor.fetchall()]

@staticmethod

def complete_task(task_id):

with get_db() as conn:

conn.execute(“””

UPDATE tasks

SET status = ‘completed’, completed_at = CURRENT_TIMESTAMP

WHERE id = ?

“””, (task_id,))

return True

@staticmethod

def delete_task(task_id):

with get_db() as conn:

conn.execute(“DELETE FROM tasks WHERE id = ?”, (task_id,))

return True

# Usage

if __name__ == “__main__”:

init_db()

# Add tasks

TaskManager.add_task(“Learn SQLite”, “Study database integration”, priority=3)

TaskManager.add_task(“Build task manager”, “Create the complete example”, priority=2)

TaskManager.add_task(“Write documentation”, priority=1)

# List tasks

tasks = TaskManager.get_all_tasks()

for task in tasks:

status_icon = “✓” if task[“status”] == “completed” else “○”

print(f”{status_icon} [{task[‘priority’]}] {task[‘title’]} – {task[‘description’]}”)

# Complete a task

TaskManager.complete_task(1)

print(“\nPending tasks:”)

pending = TaskManager.get_pending_tasks()

for task in pending:

print(f” – {task[‘title’]}”)

Common SQLite Mistakes
  • Forgetting to call `commit()` (changes are lost)
  • Using string formatting for SQL (SQL injection vulnerability)
  • Not closing connections (file locks remain)
  • Assuming SQLite supports full concurrency (write locks the database)
  • Storing dates as strings (use TIMESTAMP type)
  • Not using row_factory (accessing by index is fragile)
Check Your Understanding
  • How do you connect to an SQLite database in Python?
  • What is the purpose of `conn.commit()`?
  • Why should you use `?` placeholders instead of f-strings in SQL queries?
  • Write a function that inserts a new user into a `users` table.
  • How do you access columns by name instead of index?
  • What happens if an exception occurs inside a transaction?

⚡ Whisper

SQLite is a database in a file. No server, no configuration, no password. Just a file on your disk. But do not let its simplicity fool you. SQLite is powerful. It supports SQL. It supports transactions. It supports terabyte-sized databases. It is the most deployed database in the world. Use it for configuration, logs, user data, application state. Use it for desktop apps, mobile apps, small web apps. Learn SQL. Learn `CREATE TABLE`, `INSERT`, `SELECT`, `UPDATE`, `DELETE`. Learn `WHERE`, `JOIN`, `GROUP BY`. Learn transactions. Learn backups. And always, always use parameterized queries. SQL injection is preventable. `?` placeholders are your shield. Your data will be safe. Your queries will be fast. Your application will be persistent. The database is waiting. Store something permanent.

Related posts