SQLite3 - built -in work with SQLite

Introduction to SQLite and the sqlite3 Module

SQLite is a compact relational DBMS that does not require a separate server installation and is ideal for developing small and medium‑sized projects. In Python, SQLite support is provided through the built‑in sqlite3 module, enabling database operations without installing additional libraries.

The sqlite3 module turns database creation and usage into a process similar to working with ordinary files. This solution is especially valuable for prototyping, embedded systems, desktop applications, and educational purposes.

Installation and Compatibility

The sqlite3 module has been part of Python’s standard library since version 2.5 and is available without any extra installation steps. It offers full compatibility with Windows, macOS, and Linux, providing an interface for SQLite 3.

import sqlite3

Connecting to a Database and Creating a Connection

Creating a File‑Based Database

conn = sqlite3.connect('my_database.db')

Executing this command automatically creates the database file if it does not already exist.

Creating an In‑Memory Database

conn = sqlite3.connect(':memory:')

This option is particularly convenient for testing and temporary operations.

Additional Connection Parameters

conn = sqlite3.connect('database.db',
                      check_same_thread=False,
                      detect_types=sqlite3.PARSE_DECLTYPES)

Creating Tables and Working with the Data Schema

Creating Tables

cursor = conn.cursor()
cursor.execute('''
    CREATE TABLE users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        age INTEGER,
        email TEXT UNIQUE,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')
conn.commit()

SQLite Data Types

SQLite supports five core data types:

• INTEGER – whole numbers
• REAL – floating‑point numbers
• TEXT – character strings
• BLOB – binary data
• NULL – absence of a value

SQLite uses dynamic typing, providing flexibility but requiring careful handling of data.

CRUD Operations (Create, Read, Update, Delete)

Inserting Data

# Insert a single record
cursor.execute("INSERT INTO users (name, age, email) VALUES (?, ?, ?)",
               ("John Doe", 30, "john@example.com"))

# Insert multiple records
users_data = [
    ("Maria Smith", 25, "maria@example.com"),
    ("Peter Johnson", 35, "peter@example.com"),
    ("Anna Brown", 28, "anna@example.com")
]
cursor.executemany("INSERT INTO users (name, age, email) VALUES (?, ?, ?)", users_data)
conn.commit()

Querying Data

# Retrieve all rows
cursor.execute("SELECT * FROM users")
all_users = cursor.fetchall()

# Retrieve with a condition
cursor.execute("SELECT name, age FROM users WHERE age > ?", (25,))
adult_users = cursor.fetchall()

# Retrieve with sorting
cursor.execute("SELECT * FROM users ORDER BY age DESC")
sorted_users = cursor.fetchall()

Updating Data

cursor.execute("UPDATE users SET age = ? WHERE name = ?", (31, "John Doe"))
conn.commit()

# Get number of affected rows
print(f"Rows updated: {cursor.rowcount}")

Deleting Data

cursor.execute("DELETE FROM users WHERE name = ?", ("John Doe",))
conn.commit()

Methods for Reading Query Results

Basic Data Retrieval Methods

• fetchone() – returns a single result row
• fetchall() – returns all rows as a list
• fetchmany(size) – returns a specified number of rows

cursor.execute("SELECT * FROM users")

# Get one row
user = cursor.fetchone()
if user:
    print(f"ID: {user[0]}, Name: {user[1]}, Age: {user[2]}")

# Get several rows
users_batch = cursor.fetchmany(5)
for user in users_batch:
    print(user)

Working with Results as Dictionaries

conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")

for row in cursor.fetchall():
    print(f"Name: {row['name']}, Age: {row['age']}")

Context Managers and Transaction Management

Using Context Managers

with sqlite3.connect('my_database.db') as conn:
    cursor = conn.cursor()
    cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("Test", 25))
    # Automatic commit on success
    # Automatic rollback on error

Manual Transaction Control

conn = sqlite3.connect('database.db')
try:
    cursor = conn.cursor()
    cursor.execute("BEGIN TRANSACTION")
    cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("User1", 25))
    cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("User2", 30))
    conn.commit()
except sqlite3.Error as e:
    conn.rollback()
    print(f"Error: {e}")
finally:
    conn.close()

Creating Indexes and Constraints

Creating Indexes

# Ordinary index
cursor.execute("CREATE INDEX idx_users_age ON users(age)")

# Unique index
cursor.execute("CREATE UNIQUE INDEX idx_users_email ON users(email)")

# Composite index
cursor.execute("CREATE INDEX idx_users_name_age ON users(name, age)")

Adding Constraints

cursor.execute('''
    CREATE TABLE products (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        price REAL CHECK(price > 0),
        category_id INTEGER,
        FOREIGN KEY (category_id) REFERENCES categories(id)
    )
''')

Working with Dates and Times

Storing Dates

from datetime import datetime

now = datetime.now()
cursor.execute("INSERT INTO logs (timestamp, message) VALUES (?, ?)",
               (now, "Test message"))

Automatic Type Conversion

conn = sqlite3.connect("database.db", detect_types=sqlite3.PARSE_DECLTYPES)

# Dates are now automatically converted to datetime objects
cursor.execute("SELECT timestamp FROM logs")
for row in cursor.fetchall():
    print(type(row[0]))  # <class 'datetime.datetime'>

Importing and Exporting Data

Exporting to CSV

import csv

cursor.execute("SELECT * FROM users")
with open('users.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.writer(f)
    # Write header row
    writer.writerow([description[0] for description in cursor.description])
    # Write data rows
    writer.writerows(cursor.fetchall())

Importing from CSV

with open('users.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        cursor.execute("INSERT INTO users (name, age, email) VALUES (?, ?, ?)",
                      (row['name'], row['age'], row['email']))
conn.commit()

Creating a Database Backup

def backup_database(source_db, backup_db):
    source = sqlite3.connect(source_db)
    backup = sqlite3.connect(backup_db)
    source.backup(backup)
    source.close()
    backup.close()

Integration with Pandas and Other Libraries

Reading Data into a DataFrame

import pandas as pd

df = pd.read_sql("SELECT * FROM users WHERE age > 25", conn)
print(df.head())

# Reading with parameters
df = pd.read_sql("SELECT * FROM users WHERE age > ?", conn, params=(25,))

Saving a DataFrame to the Database

df.to_sql("new_users", conn, index=False, if_exists="replace")

# Append to an existing table
df.to_sql("users", conn, index=False, if_exists="append")

Testing with SQLite

Creating a Test Database

import unittest

class TestDatabase(unittest.TestCase):
    def setUp(self):
        self.conn = sqlite3.connect(":memory:")
        self.cursor = self.conn.cursor()
        self.cursor.execute('''
            CREATE TABLE test_users (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL
            )
        ''')
    
    def tearDown(self):
        self.conn.close()
    
    def test_insert_user(self):
        self.cursor.execute("INSERT INTO test_users (name) VALUES (?)", ("Test",))
        self.conn.commit()
        
        self.cursor.execute("SELECT COUNT(*) FROM test_users")
        count = self.cursor.fetchone()[0]
        self.assertEqual(count, 1)

Advanced sqlite3 Features

User‑Defined Functions

def reverse_string(s):
    return s[::-1]

conn.create_function("reverse", 1, reverse_string)
cursor.execute("SELECT reverse(name) FROM users")

User‑Defined Aggregate Functions

class Average:
    def __init__(self):
        self.count = 0
        self.sum = 0
    
    def step(self, value):
        self.count += 1
        self.sum += value
    
    def finalize(self):
        return self.sum / self.count if self.count > 0 else 0

conn.create_aggregate("avg", 1, Average)

Performance Optimization

Using PRAGMA Statements

cursor.execute("PRAGMA synchronous = OFF")
cursor.execute("PRAGMA journal_mode = WAL")
cursor.execute("PRAGMA cache_size = 10000")
cursor.execute("PRAGMA temp_store = MEMORY")

Bulk Data Processing

# Faster insertion of large data sets
cursor.execute("BEGIN TRANSACTION")
for i in range(10000):
    cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)",
                  (f"User{i}", i % 100))
cursor.execute("COMMIT")

Table of sqlite3 Module Methods and Functions

Security and Best Practices

Protection Against SQL Injection

# Correct – use parameterized queries
cursor.execute("SELECT * FROM users WHERE name = ?", (user_input,))

# Incorrect – avoid string formatting
# cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")

Resource Management

# Always close connections
try:
    conn = sqlite3.connect('database.db')
    # work with the database
finally:
    conn.close()

# Or use a context manager
with sqlite3.connect('database.db') as conn:
    # work with the database

Error Handling

try:
    cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("Ivan", 30))
    conn.commit()
except sqlite3.IntegrityError:
    print("Data integrity error")
except sqlite3.OperationalError:
    print("Operational error")
except sqlite3.Error as e:
    print(f"SQLite error: {e}")

Comparison with Other DBMS

Practical Usage Examples

Logging System

def create_log_table():
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS logs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
            level TEXT NOT NULL,
            message TEXT NOT NULL,
            module TEXT
        )
    ''')
    conn.commit()

def log_message(level, message, module=None):
    cursor.execute("INSERT INTO logs (level, message, module) VALUES (?, ?, ?)",
                  (level, message, module))
    conn.commit()

Data Caching

class SQLiteCache:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS cache (
                key TEXT PRIMARY KEY,
                value TEXT,
                expires DATETIME
            )
        ''')
    
    def set(self, key, value, ttl=3600):
        expires = datetime.now() + timedelta(seconds=ttl)
        self.cursor.execute("INSERT OR REPLACE INTO cache VALUES (?, ?, ?)",
                          (key, value, expires))
        self.conn.commit()
    
    def get(self, key):
        self.cursor.execute("SELECT value FROM cache WHERE key = ? AND expires > ?",
                          (key, datetime.now()))
        result = self.cursor.fetchone()
        return result[0] if result else None

Frequently Asked Questions

What is sqlite3 in Python?

sqlite3 is the built‑in Python module for working with SQLite databases. It provides an interface for creating, reading, updating, and deleting data in SQLite databases without requiring any additional software.

Do I need to install SQLite separately?

No. The sqlite3 module has been part of the Python standard library since version 2.5, and the SQLite engine itself is bundled with Python, so no extra installation is needed.

Is SQLite suitable for production?

SQLite works great for small to medium applications, desktop programs, embedded systems, and mobile apps. For large‑scale web applications with high traffic, a server‑based DBMS such as PostgreSQL or MySQL is recommended.

How can I protect against SQL injection?

Always use parameterized queries with placeholders instead of string formatting. Use the ? placeholder and pass parameters separately.

Can SQLite be used with multithreading?

Yes, but with limitations. SQLite supports multithreading, yet only one write operation can occur at a time. Enabling WAL mode improves concurrency.

How do I work with SQLite and Pandas?

Pandas offers read_sql() for fetching data from a database and to_sql() for writing a DataFrame back to SQLite. This makes integration straightforward for data analysis tasks.

How do I back up an SQLite database?

Use the backup() method or simply copy the database file. For programmatic backups, the iterdump() method can also be employed.

What limitations does SQLite have?

Key limitations include limited concurrent write support, restricted schema‑alteration capabilities, and the absence of some advanced SQL features found in server‑based DBMSs.

How can I optimize SQLite performance?

Use indexes on frequently queried columns, apply PRAGMA settings for performance tuning, wrap bulk operations in transactions, and enable WAL mode for better multithreaded performance.

Is SQLite viable for web applications?

Yes, SQLite is suitable for small web apps and prototypes. However, for high‑load applications with many simultaneous users, a server‑based DBMS is advisable.

Conclusion

The sqlite3 module turns Python into a powerful tool for database work without the need for additional software installations. With built‑in SQLite support, developers can quickly prototype, build desktop applications, and create automation systems.

SQLite combines ease of use, reliability, and strong performance, making it an excellent choice for a wide range of tasks—from programming education to commercial application development. Understanding SQLite’s capabilities and constraints helps you make an informed decision about the right database solution for any project.