def fetch_users_by_age(min_age: int, max_age: int) -> List[dict]:
"""Fixed: Uses placeholders instead of f-strings"""
query = """
SELECT id, name, email, age
FROM users
WHERE age BETWEEN ? AND ?
ORDER BY age DESC
"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute(query, (min_age, max_age))
return [dict(row) for row in cursor.fetchall()]
def add_user(name, email):
with db_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"INSERT INTO users (name, email) VALUES (?, ?)",
(name, email)
)
return cursor.lastrowid
This is the most critical section. Never use Python string concatenation for queries.
The "Wrong" Way (Vulnerable to Injection): sqlite3 tutorial query python fixed
# DANGEROUS - Do not use
name = "Robert'); DROP TABLE employees; --"
cursor.execute(f"INSERT INTO employees (name) VALUES ('name')")
The "Fixed" Way (Parameterized Query):
Use ? as placeholders. SQLite3 handles the escaping and type conversion safely.
def add_employee(name, position, salary):
try:
with sqlite3.connect('company.db') as conn:
cursor = conn.cursor()
# Use ? placeholders for security
query = "INSERT INTO employees (name, position, salary) VALUES (?, ?, ?)"
# Data must be passed as a tuple (note the comma for single items)
data = (name, position, salary)
cursor.execute(query, data)
conn.commit()
print(f"Employee name added successfully.")
except sqlite3.Error as e:
print(f"Error adding employee: e")
cursor.execute("UPDATE users SET age = 31 WHERE name = 'Alice'")
cursor.execute("DELETE FROM users WHERE email = 'alice@example.com'")
conn.commit()
# DANGEROUS - DO NOT DO THIS
cursor.execute(f"INSERT INTO users VALUES ('name')")
If name is "'; DROP TABLE users; --", you lose everything. The "Fixed" Way (Parameterized Query):
Use
import sqlite3
from datetime import datetime
class DatabaseManager:
def init(self, db_name='my_database.db'):
self.db_name = db_name
self.create_tables()
def create_tables(self):
with sqlite3.connect(self.db_name) as conn:
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
email TEXT UNIQUE NOT NULL,
age INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
def add_user(self, username, email, age):
try:
with sqlite3.connect(self.db_name) as conn:
cursor = conn.cursor()
cursor.execute(
"INSERT INTO users (username, email, age) VALUES (?, ?, ?)",
(username, email, age)
)
return cursor.lastrowid
except sqlite3.IntegrityError:
print(f"User username already exists")
return None
def get_all_users(self):
with sqlite3.connect(self.db_name) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM users ORDER BY created_at DESC")
return [dict(row) for row in cursor.fetchall()]
def update_user_age(self, username, new_age):
with sqlite3.connect(self.db_name) as conn:
cursor = conn.cursor()
cursor.execute(
"UPDATE users SET age = ? WHERE username = ?",
(new_age, username)
)
return cursor.rowcount > 0
def delete_user(self, username):
with sqlite3.connect(self.db_name) as conn:
cursor = conn.cursor()
cursor.execute("DELETE FROM users WHERE username = ?", (username,))
return cursor.rowcount > 0
As Pythonia concluded her quest, she closed the connection to the database, ensuring that her changes were saved. As Pythonia concluded her quest
conn.close()
The people of Codearia celebrated Pythonia's mastery of SQLite3, and her legendary adventures were etched into the annals of database history.
# Update a user's email
cursor.execute('UPDATE users SET email = ? WHERE id = ?', ('bob2@example.com', 3,))
conn.commit()