SQLite
Import packages
| import sqlite3
from typing import Dict, List, Tuple, Union
|
Usage
Creating a database
Creating a connection
| conn = sqlite3.connect(":memory:")
print(conn)
|
Creating a cursor
Creating a table
| create_employee_table_statement = """
CREATE TABLE IF NOT EXISTS employees(
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
age INTEGER,
department text
);
"""
cursor.execute(create_employee_table_statement)
# commit changes
conn.commit()
|
| cursor.execute("PRAGMA table_info(employees)")
for employees_col in cursor.fetchall():
print(employees_col)
|
(0, 'id', 'INTEGER', 0, None, 1)
(1, 'name', 'TEXT', 1, None, 0)
(2, 'age', 'INTEGER', 0, None, 0)
(3, 'department', 'TEXT', 0, None, 0)
Inserting data into a table
With execute
| insert_employee_table_statement = """
INSERT INTO employees(name, age, department)
VALUES
('John Doe', 33, 'Machine Learning Engineering'),
("Jane Smith", 29, 'Data Science'),
("Alice Brown", 40, 'HR')
"""
cursor.execute(insert_employee_table_statement)
# commit changes
conn.commit()
|
With execute many.
| employees = [
("Michael Johnson", 45, "Finance"),
("Emily Davis", 27, "Marketing"),
("Robert Wilson", 38, "Software Engineering"),
]
cursor.executemany(
"INSERT INTO employees (name, age, department) VALUES (?, ?, ?)", employees
)
# commit changes
conn.commit()
|
Query data
| cursor.execute("SELECT * FROM employees")
employee_rows: List[Tuple[int, str, int, str]] = cursor.fetchall()
# Get column names
column_names = [description[0] for description in cursor.description]
# Get rows
for employee_row in employee_rows:
print(dict(zip(column_names, employee_row)))
|
{'id': 1, 'name': 'John Doe', 'age': 33, 'department': 'Machine Learning Engineering'}
{'id': 2, 'name': 'Jane Smith', 'age': 29, 'department': 'Data Science'}
{'id': 3, 'name': 'Alice Brown', 'age': 40, 'department': 'HR'}
{'id': 4, 'name': 'Michael Johnson', 'age': 45, 'department': 'Finance'}
{'id': 5, 'name': 'Emily Davis', 'age': 27, 'department': 'Marketing'}
{'id': 6, 'name': 'Robert Wilson', 'age': 38, 'department': 'Software Engineering'}
Update table data
| update_employee_statement = """
UPDATE employees
SET age = ?, department = ?
WHERE id = ?
"""
cursor.execute(update_employee_statement, (30, "AI Research", 2))
# commit changes
conn.commit()
|
Check for the updated data.
| cursor.execute("SELECT * FROM employees WHERE id = 2")
updated_employee_row = cursor.fetchone()
updated_employee_row
|
(2, 'Jane Smith', 30, 'AI Research')
Delete data from table
| delete_employee_statement = """
DELETE FROM employees WHERE id = ?
"""
cursor.execute(delete_employee_statement, (2,))
# commit changes
conn.commit()
|
Check for deleted data.
| cursor.execute("SELECT * FROM employees WHERE id = 2")
deleted_employee_row = cursor.fetchone()
type(deleted_employee_row)
|
NoneType
Make rows behave like dictionaries
| conn.row_factory = sqlite3.Row
cursor = conn.cursor()
|
| cursor.execute("SELECT * FROM employees")
employee_rows: List[Dict[str, Union[str, int]]] = cursor.fetchall()
# Get rows
for employee_row in employee_rows:
print({**employee_row})
|
{'id': 1, 'name': 'John Doe', 'age': 33, 'department': 'Machine Learning Engineering'}
{'id': 3, 'name': 'Alice Brown', 'age': 40, 'department': 'HR'}
{'id': 4, 'name': 'Michael Johnson', 'age': 45, 'department': 'Finance'}
{'id': 5, 'name': 'Emily Davis', 'age': 27, 'department': 'Marketing'}
{'id': 6, 'name': 'Robert Wilson', 'age': 38, 'department': 'Software Engineering'}