BigXStar Logo
Advanced Topics

Python Database Connectivity

Learn how to connect Python to databases using sqlite3, perform CRUD operations, and use SQLAlchemy ORM for robust database management.

Python Database Connectivity

Python provides excellent database support through built-in and third-party libraries. Whether you need a simple local database or enterprise-grade connectivity, Python has you covered.

SQLite with sqlite3 (Built-in)

SQLite is a lightweight, file-based database included with Python — no setup required:

python
import sqlite3

# Connect to database (creates file if not exists)
conn = sqlite3.connect("myapp.db")

# Create a cursor
cursor = conn.cursor()

# Create a table
cursor.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        age INTEGER,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

conn.commit()
conn.close()

Always use context managers or explicitly close connections to prevent data loss.

CRUD Operations

Create (INSERT)

python
import sqlite3

conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()

# Insert a single record (use ? placeholders to prevent SQL injection)
cursor.execute(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
    ("Alice", "alice@example.com", 30)
)

# Insert multiple records
users = [
    ("Bob", "bob@example.com", 25),
    ("Charlie", "charlie@example.com", 35),
    ("Diana", "diana@example.com", 28),
]
cursor.executemany(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
    users
)

conn.commit()
print(f"Inserted {cursor.rowcount} rows")
print(f"Last row ID: {cursor.lastrowid}")
conn.close()

Never use f-strings or string formatting in SQL queries — always use parameterized queries (? placeholders) to prevent SQL injection.

Read (SELECT)

python
import sqlite3

conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()

# Fetch all rows
cursor.execute("SELECT * FROM users")
all_users = cursor.fetchall()
for user in all_users:
    print(user)  # Tuple: (1, 'Alice', 'alice@example.com', 30, '2024-...')

# Fetch one row
cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
user = cursor.fetchone()
print(user)

# Fetch with conditions
cursor.execute("SELECT name, age FROM users WHERE age > ?", (25,))
results = cursor.fetchall()
for name, age in results:
    print(f"{name}: {age} years old")

# Use Row factory for dict-like access
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
user = cursor.fetchone()
print(user["name"])    # Alice
print(user["email"])   # alice@example.com
print(dict(user))      # Full dictionary

conn.close()

Update

python
import sqlite3

conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()

# Update a record
cursor.execute(
    "UPDATE users SET age = ? WHERE name = ?",
    (31, "Alice")
)

conn.commit()
print(f"Updated {cursor.rowcount} rows")
conn.close()

Delete

python
import sqlite3

conn = sqlite3.connect("myapp.db")
cursor = conn.cursor()

# Delete a record
cursor.execute("DELETE FROM users WHERE id = ?", (1,))

conn.commit()
print(f"Deleted {cursor.rowcount} rows")
conn.close()

Using Context Managers

python
import sqlite3

# Automatic commit/rollback
with sqlite3.connect("myapp.db") as conn:
    cursor = conn.cursor()

    try:
        cursor.execute(
            "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
            ("Eve", "eve@example.com", 22)
        )
        # Auto-commits on success
    except sqlite3.IntegrityError as e:
        # Auto-rollbacks on error
        print(f"Database error: {e}")

Database Helper Class

python
import sqlite3
from contextlib import contextmanager


class Database:
    """Simple SQLite database wrapper."""

    def __init__(self, db_path):
        self.db_path = db_path

    @contextmanager
    def get_connection(self):
        """Get a database connection with auto-commit."""
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

    def execute(self, query, params=None):
        """Execute a query and return results."""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(query, params or ())
            return cursor.fetchall()

    def execute_one(self, query, params=None):
        """Execute a query and return one result."""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(query, params or ())
            return cursor.fetchone()

    def execute_write(self, query, params=None):
        """Execute a write query and return affected rows."""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(query, params or ())
            return cursor.rowcount


# Usage
db = Database("myapp.db")

# Query
users = db.execute("SELECT * FROM users WHERE age > ?", (25,))
for user in users:
    print(f"{user['name']}: {user['age']}")

# Insert
db.execute_write(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
    ("Frank", "frank@example.com", 40)
)

# Single result
user = db.execute_one("SELECT * FROM users WHERE id = ?", (1,))
if user:
    print(dict(user))

MySQL and PostgreSQL

For production databases, use dedicated drivers:

MySQL

python
# Install: pip install mysql-connector-python
import mysql.connector

conn = mysql.connector.connect(
    host="localhost",
    user="root",
    password="password",
    database="myapp"
)

cursor = conn.cursor(dictionary=True)
cursor.execute("SELECT * FROM users WHERE age > %s", (25,))
results = cursor.fetchall()

for row in results:
    print(f"{row['name']}: {row['email']}")

conn.close()

PostgreSQL

python
# Install: pip install psycopg2-binary
import psycopg2
from psycopg2.extras import RealDictCursor

conn = psycopg2.connect(
    host="localhost",
    dbname="myapp",
    user="postgres",
    password="password"
)

cursor = conn.cursor(cursor_factory=RealDictCursor)
cursor.execute("SELECT * FROM users WHERE age > %s", (25,))
results = cursor.fetchall()

for row in results:
    print(f"{row['name']}: {row['email']}")

conn.commit()
conn.close()

SQLAlchemy ORM

SQLAlchemy provides a powerful Object-Relational Mapping (ORM) layer:

python
# Install: pip install sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime

# Setup
engine = create_engine("sqlite:///myapp.db", echo=False)
Base = declarative_base()
Session = sessionmaker(bind=engine)


# Define model
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    email = Column(String(200), unique=True, nullable=False)
    age = Column(Integer)
    created_at = Column(DateTime, default=datetime.utcnow)

    def __repr__(self):
        return f"<User(name='{self.name}', email='{self.email}')>"

    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "email": self.email,
            "age": self.age,
        }


# Create tables
Base.metadata.create_all(engine)

ORM CRUD Operations

python
# Create
session = Session()

user = User(name="Alice", email="alice@example.com", age=30)
session.add(user)
session.commit()
print(f"Created: {user}")

# Bulk create
users = [
    User(name="Bob", email="bob@example.com", age=25),
    User(name="Charlie", email="charlie@example.com", age=35),
]
session.add_all(users)
session.commit()

# Read
all_users = session.query(User).all()
for u in all_users:
    print(u.to_dict())

# Filter
young_users = session.query(User).filter(User.age < 30).all()
alice = session.query(User).filter_by(name="Alice").first()

# Update
alice.age = 31
session.commit()

# Delete
session.delete(alice)
session.commit()

session.close()

Practical Example: Task Manager

python
"""
Simple task manager using SQLite.
"""

import sqlite3
from datetime import datetime


class TaskManager:
    def __init__(self, db_path="tasks.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS tasks (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT NOT NULL,
                    description TEXT,
                    priority TEXT DEFAULT 'medium',
                    completed BOOLEAN DEFAULT 0,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    completed_at TIMESTAMP
                )
            """)

    def add_task(self, title, description="", priority="medium"):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(
                "INSERT INTO tasks (title, description, priority) VALUES (?, ?, ?)",
                (title, description, priority)
            )
            return cursor.lastrowid

    def complete_task(self, task_id):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE tasks SET completed = 1, completed_at = ? WHERE id = ?",
                (datetime.now().isoformat(), task_id)
            )

    def get_tasks(self, show_completed=False):
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            if show_completed:
                query = "SELECT * FROM tasks ORDER BY priority, created_at"
            else:
                query = "SELECT * FROM tasks WHERE completed = 0 ORDER BY priority, created_at"
            return conn.execute(query).fetchall()

    def delete_task(self, task_id):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))

    def show_tasks(self, show_completed=False):
        tasks = self.get_tasks(show_completed)
        if not tasks:
            print("No tasks found!")
            return

        priority_icons = {"high": "🔴", "medium": "🟡", "low": "🟢"}

        print(f"\n{'ID':<4} {'Status':<8} {'Priority':<10} {'Title'}")
        print("-" * 50)
        for task in tasks:
            status = "✅" if task["completed"] else "⬜"
            icon = priority_icons.get(task["priority"], "⚪")
            print(f"{task['id']:<4} {status:<8} {icon} {task['priority']:<7} {task['title']}")


# Usage
tm = TaskManager()

tm.add_task("Learn Python databases", "Study sqlite3 and SQLAlchemy", "high")
tm.add_task("Build API client", "Create reusable HTTP client", "medium")
tm.add_task("Write unit tests", priority="low")

tm.show_tasks()
tm.complete_task(1)
tm.show_tasks(show_completed=True)

Summary

  • SQLite (sqlite3) is built into Python — great for development and small apps
  • Always use parameterized queries (? placeholders) to prevent SQL injection
  • Use context managers for automatic connection management
  • row_factory = sqlite3.Row enables dict-like access to results
  • MySQL: use mysql-connector-python; PostgreSQL: use psycopg2
  • SQLAlchemy ORM provides a Pythonic way to interact with any database
  • Build wrapper classes for clean, reusable database code
  • Always commit after write operations and close connections

Congratulations! You've completed all the core Python topics. Continue practicing by building projects and exploring Python's vast ecosystem of libraries!

PreviousWorking with APIs in PythonAdvanced Topics
← Back to Learn Python Overview