#================================================================================#
# Cartonnage has been tested on the following database versions:
#
# SQLite3 version 3.51.0
# Oracle 19c, 23ai, and 26ai
# MySQL 8.0.35
# PostgreSQL 17.7
# MicrosoftAzureSQL NA
#================================================================================#
import sqlite3
# Create database connection
connection = sqlite3.connect('hr.db', check_same_thread=True)
# Set the database adapter for Record class
# SQLite3 can be replaced by Oracle, MySQL, Postgres, or MicrosoftSQL
Record.database__ = SQLite3(connection)
# Transaction control - commit or rollback manually
Record.database__.commit() # Commit changes
Record.database__.rollback() # Rollback changes
Record.database__.close() # Close connection
# No SQL generated - connection setup only
# Supported databases:
# - SQLite3(connection)
# - Oracle(connection)
# - MySQL(connection)
# - Postgres(connection)
# - MicrosoftSQL(connection)
# No parameters - connection setup
# Database connection established
# Transaction control available via:
# - Record.database__.commit()
# - Record.database__.rollback()
# - Record.database__.close()
# Table Record classes
class Regions(Record): pass
class Countries(Record): pass
class Departments(Record): pass
class Jobs(Record): pass
class Job_History(Record): pass
class Locations(Record): pass
class Employees(Record): pass
class Dependents(Record): pass
# Alias classes for self-joins and subqueries
class RegionsAlias(Regions): pass
class CountriesAlias(Countries): pass
class DepartmentsAlias(Departments): pass
class JobsAlias(Jobs): pass
class Job_HistoryAlias(Job_History): pass
class LocationsAlias(Locations): pass
class EmployeesAlias(Employees): pass
class DependentsAlias(Dependents): pass
# Self-join alias for employee-manager relationship
class Managers(Employees): pass
# No SQL generated - class definitions only
# Classes map to tables:
# Regions -> regions
# Countries -> countries
# Departments -> departments
# Jobs -> jobs
# Job_History -> job_history
# Locations -> locations
# Employees -> employees
# Dependents -> dependents
# Managers -> employees (alias)
# No parameters - class definitions
# Classes are now available for ORM operations:
# - Direct table access (Employees, Jobs, etc.)
# - Alias classes for subqueries (EmployeesAlias, etc.)
# - Self-join support (Managers extends Employees)
# filter using Record.field = exact value
employee = Employees().value(employee_id=100).select()
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert employee.data == {'employee_id': 100, 'first_name': 'Steven', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 24000, 'commission_pct': None, 'manager_id': None, 'department_id': 9}
SELECT * FROM Employees Employees
WHERE Employees.employee_id = ?
[100]
# filter using Record.where(expression)
employee = (Employees().where(Employees.employee_id == 100).select())
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert employee.data == {'employee_id': 100, 'first_name': 'Steven', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 24000, 'commission_pct': None, 'manager_id': None, 'department_id': 9}
SELECT * FROM Employees Employees
WHERE Employees.employee_id = ?
[100]
# join two tables
employee = (
Employees()
.join(Dependents, (Employees.employee_id == Dependents.employee_id))
.where(Dependents.first_name == "Jennifer")
.select(selected="Employees.*")
)
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert employee.data == {'employee_id': 100, 'first_name': 'Steven', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 24000, 'commission_pct': None, 'manager_id': None, 'department_id': 9}
SELECT Employees.* FROM Employees Employees
INNER JOIN Dependents Dependents ON Employees.employee_id = Dependents.employee_id
WHERE Dependents.first_name = ?
['Jennifer']
# join same table
employee = (
Employees()
.join(Managers, (Employees.manager_id == Managers.employee_id))
.where(Managers.first_name == 'Lex')
.select(selected='Employees.*, Managers.employee_id AS "manager_employee_id", Managers.email AS "manager_email"') # selected columns
)
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert employee.data == {'employee_id': 103, 'first_name': 'Alexander', 'last_name': 'Hunold', 'email': 'alexander.hunold@sqltutorial.org', 'phone_number': '590.423.4567', 'hire_date': '1990-01-03', 'job_id': 9, 'salary': 9000, 'commission_pct': None, 'manager_id': 102, 'department_id': 6, 'manager_employee_id': 102, 'manager_email': 'lex.de haan@sqltutorial.org'}
SELECT Employees.*, Managers.employee_id AS "manager_employee_id", Managers.email AS "manager_email"
FROM Employees Employees
INNER JOIN Employees Managers ON Employees.manager_id = Managers.employee_id
WHERE Managers.first_name = ?
['Lex']
# filter using like() # use two diffent filteration methods
employee = (
Employees()
.like(first_name = 'Stev%') # calls internally Record.filter_.like()
.where(Employees.first_name.like('Stev%')) # calls internally Record.filter_.like()
.select()
)
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert employee.data == {'employee_id': 100, 'first_name': 'Steven', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 24000, 'commission_pct': None, 'manager_id': None, 'department_id': 9}
SELECT * FROM Employees Employees
WHERE Employees.first_name LIKE ?
AND Employees.first_name LIKE ?
['Stev%', 'Stev%']
# filter using is_null() # use two diffent filteration methods
employee = (
Employees()
.is_null(manager_id = None) # calls internally Record.filter_.is_null()
.where(Employees.manager_id.is_null()) # calls internally Record.filter_.where()
.select()
)
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert employee.data == {'employee_id': 100, 'first_name': 'Steven', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 24000, 'commission_pct': None, 'manager_id': None, 'department_id': 9}
SELECT * FROM Employees Employees
WHERE Employees.manager_id IS NULL
AND Employees.manager_id IS NULL
[]
# filter using is_not_null() # use two diffent filteration methods
employee = (
Employees()
.where(employee_id = 101)
.is_not_null(manager_id = None) # calls internally Record.filter_.is_not_null()
.where(Employees.manager_id.is_not_null()) # calls internally Record.filter_.where()
.select()
)
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert employee.data == {'employee_id': 101, 'first_name': 'Neena', 'last_name': 'Kochhar', 'email': 'neena.kochhar@sqltutorial.org', 'phone_number': '515.123.4568', 'hire_date': '1989-09-21', 'job_id': 5, 'salary': 17000, 'commission_pct': None, 'manager_id': 100, 'department_id': 9}
SELECT * FROM Employees Employees
WHERE Employees.employee_id = ?
AND Employees.manager_id IS NOT NULL
AND Employees.manager_id IS NOT NULL
[101]
# filter using in_() # use two diffent filteration methods
employee = (
Employees()
.in_(employee_id = [100]) # calls internally Record.filter_.in_()
.where(Employees.employee_id.in_([100])) # calls internally Record.filter_.where()
.select()
)
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert employee.data == {'employee_id': 100, 'first_name': 'Steven', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 24000, 'commission_pct': None, 'manager_id': None, 'department_id': 9}
SELECT * FROM Employees Employees
WHERE Employees.employee_id IN (?)
AND Employees.employee_id IN (?)
[100, 100]
# filter using not_in() # use two diffent filteration methods
employee = (
Employees()
.where(Employees.first_name.like('Alex%'))
.not_in(employee_id = [115]) # calls internally Record.filter_.like()
.where(Employees.employee_id.not_in([115])) # calls internally Record.filter_.where()
.select()
)
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert employee.data == {'employee_id': 103, 'first_name': 'Alexander', 'last_name': 'Hunold', 'email': 'alexander.hunold@sqltutorial.org', 'phone_number': '590.423.4567', 'hire_date': '1990-01-03', 'job_id': 9, 'salary': 9000, 'commission_pct': None, 'manager_id': 102, 'department_id': 6}
SELECT * FROM Employees Employees
WHERE Employees.first_name LIKE ?
AND Employees.employee_id NOT IN (?)
AND Employees.employee_id NOT IN (?)
['Alex%', 115, 115]
# filter using between() # use two diffent filteration methods
employee = (
Employees()
.between(employee_id = (100, 101)) # calls internally Record.filter_.between()
.where(Employees.employee_id.between(100, 101)) # calls internally Record.filter_.where()
.select()
)
for e in employee:
e.hire_date = str(e.hire_date)[:10] # convert datetime to str
e.salary = float(e.salary)
assert employee.recordset.data == [
{'employee_id': 100, 'first_name': 'Steven', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 24000, 'commission_pct': None, 'manager_id': None, 'department_id': 9},
{'employee_id': 101, 'first_name': 'Neena', 'last_name': 'Kochhar', 'email': 'neena.kochhar@sqltutorial.org', 'phone_number': '515.123.4568', 'hire_date': '1989-09-21', 'job_id': 5, 'salary': 17000, 'commission_pct': None, 'manager_id': 100, 'department_id': 9}
]
SELECT * FROM Employees Employees
WHERE Employees.employee_id BETWEEN ? AND ?
AND Employees.employee_id BETWEEN ? AND ?
[100, 101, 100, 101]
# filter using gt(), ge(), le(), and lt() # use two diffent filteration methods
# use filter chaining # use & expressions
employee = (
Employees()
.gt(employee_id = 99).ge(employee_id = 100).le(employee_id = 101).lt(employee_id = 102) # calls internally Record.filter_.XY()
.where(
(Employees.employee_id > 99) &
(Employees.employee_id >= 100) &
(Employees.employee_id <= 101) &
(Employees.employee_id < 102)
) # calls internally Record.filter_.where()
.select()
)
for e in employee:
e.hire_date = str(e.hire_date)[:10] # convert datetime to str
e.salary = float(e.salary)
assert employee.recordset.data == [
{'employee_id': 100, 'first_name': 'Steven', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 24000, 'commission_pct': None, 'manager_id': None, 'department_id': 9},
{'employee_id': 101, 'first_name': 'Neena', 'last_name': 'Kochhar', 'email': 'neena.kochhar@sqltutorial.org', 'phone_number': '515.123.4568', 'hire_date': '1989-09-21', 'job_id': 5, 'salary': 17000, 'commission_pct': None, 'manager_id': 100, 'department_id': 9}
]
SELECT * FROM Employees Employees
WHERE Employees.employee_id > ?
AND Employees.employee_id >= ?
AND Employees.employee_id <= ?
AND Employees.employee_id < ?
AND (((Employees.employee_id > ? AND Employees.employee_id >= ?) AND Employees.employee_id <= ?) AND Employees.employee_id < ?)
[99, 100, 101, 102, 99, 100, 101, 102]
# & | filters and expressions
f1 = (
((Employees.employee_id == 100) & (Employees.first_name == "Steven")) |
((Employees.employee_id == 101) & (Employees.first_name == "Neena"))
)
f2 = (
((Employees.employee_id == 102) & (Employees.first_name == "Lex")) |
((Employees.employee_id == 103) & (Employees.first_name == "Alexander"))
)
employee = (
Employees()
.where(f1 | f2)
.select()
)
for e in employee:
e.hire_date = str(e.hire_date)[:10] # convert datetime to str
e.salary = float(e.salary)
assert employee.recordset.data == [
{'employee_id': 100, 'first_name': 'Steven', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 24000, 'commission_pct': None, 'manager_id': None, 'department_id': 9},
{'employee_id': 101, 'first_name': 'Neena', 'last_name': 'Kochhar', 'email': 'neena.kochhar@sqltutorial.org', 'phone_number': '515.123.4568', 'hire_date': '1989-09-21', 'job_id': 5, 'salary': 17000, 'commission_pct': None, 'manager_id': 100, 'department_id': 9},
{'employee_id': 102, 'first_name': 'Lex', 'last_name': 'De Haan', 'email': 'lex.de haan@sqltutorial.org', 'phone_number': '515.123.4569', 'hire_date': '1993-01-13', 'job_id': 5, 'salary': 17000, 'commission_pct': None, 'manager_id': 100, 'department_id': 9},
{'employee_id': 103, 'first_name': 'Alexander', 'last_name': 'Hunold', 'email': 'alexander.hunold@sqltutorial.org', 'phone_number': '590.423.4567', 'hire_date': '1990-01-03', 'job_id': 9, 'salary': 9000, 'commission_pct': None, 'manager_id': 102, 'department_id': 6}
]
SELECT * FROM Employees Employees
WHERE (((Employees.employee_id = ? AND Employees.first_name = ?) OR (Employees.employee_id = ? AND Employees.first_name = ?)) OR ((Employees.employee_id = ? AND Employees.first_name = ?) OR (Employees.employee_id = ? AND Employees.first_name = ?)))
[100, 'Steven', 101, 'Neena', 102, 'Lex', 103, 'Alexander']
# filter using multiple kwargs # use two diffent filteration methods
employee = (
Employees()
.like(first_name = 'Stev%', last_name = 'Ki%') # calls internally Record.filter_.like()
.select()
)
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert employee.data == {'employee_id': 100, 'first_name': 'Steven', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 24000, 'commission_pct': None, 'manager_id': None, 'department_id': 9}
SELECT * FROM Employees Employees
WHERE Employees.first_name LIKE ?
AND Employees.last_name LIKE ?
['Stev%', 'Ki%']
# select all and get recordset [all records] count, convert them to list of Dictionaries/lists
reg = Regions().all()
assert reg.recordset.count() == 4
assert reg.columns == ['region_id', 'region_name']
assert reg.recordset.toDicts() == [{'region_id': 1, 'region_name': 'Europe'}, {'region_id': 2, 'region_name': 'Americas'}, {'region_id': 3, 'region_name': 'Asia'}, {'region_id': 4, 'region_name': 'Middle East and Africa'}]
assert reg.recordset.toLists() == [[1, 'Europe'], [2, 'Americas'], [3, 'Asia'], [4, 'Middle East and Africa']]
SELECT * FROM Regions Regions
[]
# insert single record
emp1 = Employees()
emp1.data = {'employee_id': 19950519, 'first_name': 'William', 'last_name': 'Wallace', 'email': None, 'phone_number': '555.666.777'}
emp1.insert()
# assert emp1.rowsCount() == 1 # confirm number of inserted rows
INSERT INTO Employees (employee_id, first_name, last_name, phone_number) VALUES (?, ?, ?, ?)
[19950519, 'William', 'Wallace', '555.666.777']
# update single record
emp1 = (
Employees()
.value(employee_id=19950519)
.set(email='william.wallace@sqltutorial.org', phone_number=None)
.update()
)
# assert emp1.rowsCount() == 1 # confirm number of updated rows
UPDATE Employees SET email=?, phone_number=?
WHERE Employees.employee_id = ?
['william.wallace@sqltutorial.org', None, 19950519]
# check updated record
emp1 = Employees()
emp1.value(employee_id=19950519).select()
employee.hire_date = str(employee.hire_date)[:10] # convert datetime to str
employee.salary = float(employee.salary)
assert emp1.data == {'employee_id': 19950519, 'first_name': 'William', 'last_name': 'Wallace', 'email': 'william.wallace@sqltutorial.org', 'phone_number': None, 'hire_date': None, 'job_id': None, 'salary': None, 'commission_pct': None, 'manager_id': None, 'department_id': None}
SELECT * FROM Employees Employees
WHERE Employees.employee_id = ?
[19950519]
# delete by exists and subquery
emp1 = (
EmployeesAlias()
.where(
(Dependents.employee_id == EmployeesAlias.employee_id) &
(EmployeesAlias.email == 'william.gietz@sqltutorial.org') &
(EmployeesAlias.employee_id.in_([206]))
)
)
emp2 = Employees().where(Employees.manager_id == 205)
dep1 = (
Dependents()
.exists( _ = emp1)
.in_subquery(employee_id = emp2, selected="employee_id")
.where(Dependents._.exists(emp1))
.where(Dependents.employee_id.in_subquery(emp2, selected="employee_id"))
.delete()
)
assert dep1.rowsCount() == 1
dep1 = Dependents()
dep1.where(Dependents.employee_id == 206).select()
assert dep1.data == {}
DELETE FROM Dependents
WHERE EXISTS (
SELECT 1 FROM Employees EmployeesAlias
WHERE ((Dependents.employee_id = EmployeesAlias.employee_id AND EmployeesAlias.email = ?) AND EmployeesAlias.employee_id IN (?))
) AND Dependents.employee_id IN (
SELECT employee_id FROM Employees Employees
WHERE Employees.manager_id = ?
) AND EXISTS (
SELECT 1 FROM Employees EmployeesAlias
WHERE ((Dependents.employee_id = EmployeesAlias.employee_id AND EmployeesAlias.email = ?) AND EmployeesAlias.employee_id IN (?))
) AND Dependents.employee_id IN (
SELECT employee_id FROM Employees Employees
WHERE Employees.manager_id = ?
)
['william.gietz@sqltutorial.org', 206, 205, 'william.gietz@sqltutorial.org', 206, 205]
# insert single record
try:
emp = (
Employees()
.where(Employees.first_name == 'Steve')
.value(employee_id=1000, first_name='Ahmed', last_name='ELSamman')
.insert()
)
except Exception as e:
# will raise and exception because you added where to an insert statement
print(e)
Record.database__.rollback() # Force rollback
# Will raise an exception - WHERE clause is not valid for INSERT
[]
# filter and fetchmany records into recordset
jobs = Jobs().where(Jobs.job_title.like('%Accountant%')).select()
assert jobs.recordset.count() == 2
assert jobs.columns == ['job_id', 'job_title', 'min_salary', 'max_salary']
assert jobs.recordset.toLists() == [[1, 'Public Accountant', 4200, 9000], [6, 'Accountant', 4200, 9000]]
assert jobs.recordset.toDicts() == [{'job_id': 1, 'job_title': 'Public Accountant', 'min_salary': 4200, 'max_salary': 9000}, {'job_id': 6, 'job_title': 'Accountant', 'min_salary': 4200, 'max_salary': 9000}]
print("----------04A----------")
# iterate over recordset and update records one by one: (not recommended if you can update with one predicate)
for job in jobs:
job.set(min_salary=4500)
jobs.recordset.set(min_salary=5000)
for job in jobs.recordset: # iterate over recordset
assert job.job_id in [1,6], job.job_id
jobs.recordset.update()
recordsetLen = len(jobs.recordset) # get len() of a Recordset
assert recordsetLen == 2, recordsetLen
### it works because no field in any record of the recordset's records is set to Null.
### but if you are not sure that if your recordset's records have a Null value you have to set the onColumns parameter.
# jobs.recordset.update(onColumns=['job_id'])
jobs = Jobs().where(Jobs.job_title.like('%Accountant%')).select()
assert jobs.recordset.toLists() == [[1, 'Public Accountant', 5000, 9000], [6, 'Accountant', 5000, 9000]], jobs.recordset.toLists() # confirm recordset update
secondJobInRecordset = jobs.recordset[1] # access record instance by index
assert secondJobInRecordset.job_id == 6, secondJobInRecordset.job_id
print("----------04B----------")
jobs.recordset.delete()
### it works because no field in any record of the recordset's records is set to Null.
### but if you are not sure that if your recordset's records have a Null value you have to set the onColumns parameter.
# jobs.recordset.delete(onColumns=['job_id'])
print("----------04C----------")
jobs = Jobs().where(Jobs.job_title.like('%Accountant%')).select()
assert jobs.recordset.toLists() == [] # confirm recordset delete
Record.database__.rollback() # Force rollback
# SELECT
SELECT * FROM Jobs Jobs WHERE Jobs.job_title LIKE ?
# UPDATE (for each record in recordset)
UPDATE Jobs SET min_salary=?
WHERE Jobs.job_id = ? AND Jobs.job_title = ? AND Jobs.min_salary = ? AND Jobs.max_salary = ? AND Jobs.job_title LIKE ?
# DELETE (for each record in recordset)
DELETE FROM Jobs
WHERE Jobs.job_id = ? AND Jobs.job_title = ? AND Jobs.min_salary = ? AND Jobs.max_salary = ? AND Jobs.job_title LIKE ?
# SELECT
['%Accountant%']
# UPDATE
[(5000, 1, 'Public Accountant', 4200, 9000, '%Accountant%'), (5000, 6, 'Accountant', 4200, 9000, '%Accountant%')]
# DELETE
[(1, 'Public Accountant', 5000, 9000, '%Accountant%'), (6, 'Accountant', 5000, 9000, '%Accountant%')]
# recordset from list of dicts
recordset = Recordset.fromDicts(Employees,
[
{'employee_id': 5, 'first_name': "Mickey", 'last_name': "Mouse"},
{'employee_id': 6, 'first_name': "Donald", 'last_name': "Duck"}
]
)
assert recordset.data == [{'employee_id': 5, 'first_name': 'Mickey', 'last_name': 'Mouse'}, {'employee_id': 6, 'first_name': 'Donald', 'last_name': 'Duck'}], recordset.toDicts()
# instantiate recordset
recordset = Recordset()
# create employee(s)/record(s) instances
e1 = Employees().value(employee_id=5, first_name="Mickey", last_name="Mouse")
e2 = Employees().value(employee_id=6, first_name="Donald", last_name="Duck")
# add employee(s)/record(s) instances to the previously instantiated Recordset
recordset.add(e1, e2)
# Recordset insert:
recordset.insert()
if Record.database__.name == "MicrosoftSQL":
pass
else:
assert recordset.rowsCount() == 2, recordset.rowsCount() # not work for Azure/MicrosoftSQL only SQlite3/Oracle/MySQL/Postgres
e1.set(manager_id=77)
e2.set(manager_id=88)
e1.where(Employees.employee_id > 4) # add general condition to all records of the recordset
# Recordset update:
recordset.update()
if Record.database__.name == "MicrosoftSQL":
employees = Employees().in_(manager_id = [77, 88]).select()
assert employees.recordset.toLists() == [[5, 'Mickey', 'Mouse', None, None, None, None, None, None, 77, None], [6, 'Donald', 'Duck', None, None, None, None, None, None, 88, None]]
else:
assert recordset.rowsCount() == 2 # not work for Azure/MicrosoftSQL only SQlite3/Oracle/MySQL/Postgres
# Recordset delete:
e1.where(Employees.manager_id < 100) # add general condition to all records of the recordset
recordset.delete()
if Record.database__.name == "MicrosoftSQL":
employees = Employees().in_(manager_id = [77, 88]).select()
assert employees.recordset.toLists() == []
else:
assert recordset.rowsCount() == 2 # not work for Azure/MicrosoftSQL only SQlite3/Oracle/MySQL/Postgres
Record.database__.rollback() # Force rollback
# Batch INSERT
INSERT INTO Employees (employee_id, first_name, last_name) VALUES (?, ?, ?)
# Batch UPDATE (with general condition)
UPDATE Employees SET manager_id=?
WHERE Employees.employee_id = ? AND Employees.first_name = ? AND Employees.last_name = ? AND Employees.employee_id > ?
# Batch DELETE (with general condition)
DELETE FROM Employees
WHERE Employees.employee_id = ? AND Employees.first_name = ? AND Employees.last_name = ? AND Employees.manager_id = ? AND Employees.employee_id > ? AND Employees.manager_id < ?
# INSERT
[(5, 'Mickey', 'Mouse'), (6, 'Donald', 'Duck')]
# UPDATE
[(77, 5, 'Mickey', 'Mouse', 4), (88, 6, 'Donald', 'Duck', 4)]
# DELETE
[(5, 'Mickey', 'Mouse', 77, 4, 100), (6, 'Donald', 'Duck', 88, 4, 100)]
# Execute raw sql statement and get recordset of the returned rows
records = Record(statement="SELECT * FROM Employees WHERE employee_id IN(100, 101, 102) ", operation=Database.select)
assert records.recordset.count() == 3
for record in records:
assert record.employee_id in [100, 101, 102]
SELECT * FROM Employees WHERE employee_id IN(100, 101, 102)
# No parameters - raw SQL
# Execute parameterized sql statement and get recordset of the returned rows
placeholder = Record.database__.placeholder()
records = Record(statement=f"SELECT * FROM Employees WHERE employee_id IN({placeholder}, {placeholder}, {placeholder})", parameters=(100, 101, 102), operation=Database.select)
assert records.recordset.count() == 3
for record in records:
assert record.employee_id in [100, 101, 102]
SELECT * FROM Employees WHERE employee_id IN(?, ?, ?)
[100, 101, 102]
# instantiating Class's instance with fields' values
employee = Employees(statement=None, parameters=None, employee_id=1000, first_name="Super", last_name="Man", operation=Database.select)
employee.insert()
employee = Employees().value(employee_id=1000).select()
assert employee.data == {'employee_id': 1000, 'first_name': 'Super', 'last_name': 'Man', 'email': None, 'phone_number': None, 'hire_date': None, 'job_id': None, 'salary': None, 'commission_pct': None, 'manager_id': None, 'department_id': None}
INSERT INTO Employees (employee_id, first_name, last_name) VALUES (?, ?, ?)
[1000, 'Super', 'Man']
# group_by with HAVING clause
employees = Employees().select(selected='manager_id, count(1) AS "count"', group_by='manager_id HAVING count(1) > 4', order_by='manager_id ASC')
assert employees.recordset.data == [
{'manager_id': 100, 'count': 14}, {'manager_id': 101, 'count': 5},
{'manager_id': 108, 'count': 5}, {'manager_id': 114, 'count': 5}
], employees.recordset.data
SELECT manager_id, count(1) AS "count" FROM Employees Employees
WHERE 1=1
GROUP BY manager_id HAVING count(1) > 4
ORDER BY manager_id ASC
[]
emp = Employees().where(Employees.first_name.in_(['Steven', 'Neena']))
emp1 = Employees().set(**{'employee_id': 100, 'first_name': 'Ahmed', 'salary': 4000})
emp2 = Employees().set(**{'employee_id': 101, 'first_name': 'Kamal', 'salary': 5000})
# you can also set_.new = {} directly if you are sure of the datatype(s) validation
emp1.set__.new = {'employee_id': 100, 'first_name': 'Ahmed', 'salary': 4000}
emp2.set__.new = {'employee_id': 101, 'first_name': 'Kamal', 'salary': 5000}
rs = Recordset()
rs.add(emp1, emp2)
if(Record.database__.name in ["SQLite3", "Postgres"]):
### SQLite3 and Postgres
emp1.where(
(EXCLUDED.salary < Employees.salary) &
(Employees.last_name.in_subquery(emp, selected='last_name'))
)
if(Record.database__.name in ["Oracle", "MicrosoftSQL"]):
## Oracle and MicroftSQL
emp1.where(
(S.salary < T.salary) &
(T.last_name.in_subquery(emp, selected='last_name'))
)
rs.upsert(onColumns='employee_id')
# print(emp1.query__.statement)
# print(emp1.query__.parameters)
emp = Employees().where(Employees.employee_id.in_([100,101])).select()
if(Record.database__.name == "SQLite3"):
### SQLite3
assert emp.recordset.data == [{'employee_id': 100, 'first_name': 'Ahmed', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': '1987-06-17', 'job_id': 4, 'salary': 4000, 'commission_pct': None, 'manager_id': None, 'department_id': 9}, {'employee_id': 101, 'first_name': 'Kamal', 'last_name': 'Kochhar', 'email': 'neena.kochhar@sqltutorial.org', 'phone_number': '515.123.4568', 'hire_date': '1989-09-21', 'job_id': 5, 'salary': 5000, 'commission_pct': None, 'manager_id': 100, 'department_id': 9}], emp.recordset.data
if(Record.database__.name == "Oracle"):
### Oracle
assert emp.recordset.data == [{'employee_id': 100, 'first_name': 'Ahmed', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': datetime(1987, 6, 17, 0, 0), 'job_id': 4, 'salary': 4000, 'commission_pct': None, 'manager_id': None, 'department_id': 9}, {'employee_id': 101, 'first_name': 'Kamal', 'last_name': 'Kochhar', 'email': 'neena.kochhar@sqltutorial.org', 'phone_number': '515.123.4568', 'hire_date': datetime(1989, 9, 21, 0, 0), 'job_id': 5, 'salary': 5000, 'commission_pct': None, 'manager_id': 100, 'department_id': 9}], emp.recordset.data
if(Record.database__.name in ["MySQL", "Postgres", "MicrosoftSQL"]):
### MySQL | Postgres | Microsoft AzureSQL
assert emp.recordset.data == [{'employee_id': 100, 'first_name': 'Ahmed', 'last_name': 'King', 'email': 'steven.king@sqltutorial.org', 'phone_number': '515.123.4567', 'hire_date': date(1987, 6, 17), 'job_id': 4, 'salary': Decimal('4000.00'), 'commission_pct': None, 'manager_id': None, 'department_id': 9}, {'employee_id': 101, 'first_name': 'Kamal', 'last_name': 'Kochhar', 'email': 'neena.kochhar@sqltutorial.org', 'phone_number': '515.123.4568', 'hire_date': date(1989, 9, 21), 'job_id': 5, 'salary': Decimal('5000.00'), 'commission_pct': None, 'manager_id': 100, 'department_id': 9}], emp.recordset.data
Record.database__.rollback()
# SQLite3/Postgres:
INSERT INTO Employees (employee_id, first_name, salary)
VALUES (?, ?, ?)
ON CONFLICT (employee_id)
DO UPDATE SET first_name = EXCLUDED.first_name, salary = EXCLUDED.salary
WHERE (EXCLUDED.salary < Employees.salary AND Employees.last_name IN (
SELECT last_name FROM Employees Employees
WHERE Employees.first_name IN (?, ?)
))
# Oracle/MicrosoftSQL uses MERGE statement:
MERGE INTO Employees T
USING (SELECT ? AS employee_id, ? AS first_name, ? AS salary) S
ON (T.employee_id = S.employee_id)
WHEN MATCHED AND S.salary < T.salary AND T.last_name IN (...) THEN
UPDATE SET first_name = S.first_name, salary = S.salary
WHEN NOT MATCHED THEN
INSERT (employee_id, first_name, salary) VALUES (S.employee_id, S.first_name, S.salary)
[(100, 'Ahmed', 4000, 'Steven', 'Neena'), (101, 'Kamal', 5000, 'Steven', 'Neena')]
recordset = Recordset.fromDicts(Employees,
[
{'employee_id': 5, 'first_name': "Mickey", 'last_name': "Mouse"},
{'employee_id': 6, 'first_name': "Donald", 'last_name': "Duck"}
]
)
assert recordset.data == [{'employee_id': 5, 'first_name': 'Mickey', 'last_name': 'Mouse'}, {'employee_id': 6, 'first_name': 'Donald', 'last_name': 'Duck'}], recordset.toDicts()
# Instanitiate a session with a database instance
session = Session(Record.database__)
# Set Recordset insert query to current Session
session.set(recordset.insert_())
# Update all the Recordset's records with the same phone number
recordset.set(phone_number='+201011223344')
# Set Recordset update query to current Session
session.set(recordset.update_())
# Commit current session
session.commit()
# Select inserted and updated records
insertedEmployeesAfterUpdate = (
Employees().where(Employees.employee_id.in_([5,6])).select(selected="employee_id, first_name, last_name, phone_number")
)
assert insertedEmployeesAfterUpdate.recordset.data == [{'employee_id': 5, 'first_name': 'Mickey', 'last_name': 'Mouse', 'phone_number': '+201011223344'}, {'employee_id': 6, 'first_name': 'Donald', 'last_name': 'Duck', 'phone_number': '+201011223344'}], insertedEmployeesAfterUpdate.recordset.data
# Set Recordset delete query to current Session
session.set(recordset.delete_(onColumns=["employee_id"]))
# Commit the session
session.commit()
session.savepoint("sp1")
session.set(recordset.insert_())
session.rollbackTo('sp1')
session.releaseSavepoint('sp1')
# INSERT (via session)
INSERT INTO Employees (employee_id, first_name, last_name) VALUES (?, ?, ?)
# UPDATE (via session)
UPDATE Employees SET phone_number=?
WHERE Employees.employee_id = ? AND Employees.first_name = ? AND Employees.last_name = ?
# DELETE (via session, onColumns=["employee_id"])
DELETE FROM Employees WHERE Employees.employee_id = ?
# SAVEPOINT operations
SAVEPOINT sp1
ROLLBACK TO sp1
RELEASE SAVEPOINT sp1
# INSERT
[(5, 'Mickey', 'Mouse'), (6, 'Donald', 'Duck')]
# UPDATE
[('+201011223344', 5, 'Mickey', 'Mouse'), ('+201011223344', 6, 'Donald', 'Duck')]
# DELETE
[(5,), (6,)]
# From multiple tables
(
Employees()
.from_(Departments)
.where(
(Employees.employee_id==100) &
(Employees.department_id==Departments.department_id)
)
.select(selected="Employees.*, Departments.department_name")
)
# → SELECT ... FROM Employees Employees, Departments Departments WHERE ...
SELECT Employees.*, Departments.department_name
FROM Employees Employees, Departments Departments
WHERE (Employees.employee_id = ? AND Employees.department_id = Departments.department_id)
[100]
# From multiples tables and subqueries
class EmployeesGt1000(Employees): pass
sub = Employees().where(Employees.salary > 1000).select_().alias('EmployeesGt1000')
(
Employees().from_(sub)
.where(Employees.employee_id == EmployeesGt1000.employee_id)
.select(selected="Employees.*, EmployeesGt1000.salary")
)
# → SELECT ... FROM Employees Employees, (SELECT * FROM Employees WHERE salary > 1000) AS high_paid WHERE ...
# Raw strings # not implemented
# Employees().from_("Employees e", "Departments d").select()
SELECT Employees.*, EmployeesGt1000.salary
FROM Employees Employees,
(SELECT * FROM Employees Employees WHERE Employees.salary > ?) EmployeesGt1000
WHERE Employees.employee_id = EmployeesGt1000.employee_id
[1000]
emp = (
Employees()
.set(
first_name=Expression("UPPER(first_name)")
, last_name=Expression("LOWER(last_name)")
# UPPER and LOWER are used because:
# SQLite3, Oracle, and MySQL use new values uppered and lowered before concat.
# Postgres and MySQL uses original value before UPPER and LOWER.
, email=Expression("UPPER(first_name) || '.' || LOWER(last_name) || '@test.com'")
, salary=Employees.salary + 100
, commission_pct=Expression('COALESCE(commission_pct, 1)')
)
.value(employee_id=100)
.where(
(Employees.salary == Employees.salary) &
(Employees.salary == Employees.salary + 0)
)
.update()
.select(selected="first_name, last_name, email, salary, commission_pct")
)
emp.data['salary'] = int(emp.data['salary'])
assert emp.data == {'first_name': 'STEVEN', 'last_name': 'king', 'email': 'STEVEN.king@test.com', 'salary': 24100, 'commission_pct': 1}, emp.data
UPDATE Employees SET first_name=UPPER(first_name), last_name=LOWER(last_name), email=UPPER(first_name) || '.' || LOWER(last_name) || '@test.com', salary=Employees.salary + ?, commission_pct=COALESCE(commission_pct, 1)
WHERE Employees.employee_id = ?
AND (Employees.salary = salary AND Employees.salary = salary + 0)
[100, 100]
class Hierarchy(Record): pass # for recursive CTE
class P(Employees): pass
class C(Employees): pass
class ExecutivesDepartment(Departments): pass
class AdministrationJobs(Jobs): pass
hierarchy = Hierarchy() # used as subquery ... IN (SELECT * FROM Hierarchy) # Hierarchy is Recursive CTE
# Oracle: use hints after SELECT
# 'MySQL': Not support 'Default Behavior'
# 'MicrosoftSQL': Not supported directly, Uses temp tables instead.
# ['Oracle', 'MySQL', 'MicrosoftSQL']:
cte1 = (
P()
.where(P.manager_id.is_null())
.cte(selected='/*+ INLINE */ employee_id, manager_id, first_name', materialization=None)
)
cte2 = (
C()
.join(Hierarchy, (C.manager_id == Hierarchy.employee_id))
.where(C.first_name == 'Neena')
.cte(selected='/*+ MATERIALIZE */ C.employee_id, C.manager_id, C.first_name', materialization=None)
)
cte3 = (
ExecutivesDepartment()
.where(ExecutivesDepartment.department_name == 'Executive')
.cte(selected='/*+ INLINE */ ExecutivesDepartment.*', materialization=None)
)
cte4 = (
AdministrationJobs()
.where(AdministrationJobs.job_title.like('Administration%'))
.cte(selected='/*+ MATERIALIZE */ AdministrationJobs.*', materialization=None)
)
if Record.database__.name in ['SQLite3', 'Postgres']:
cte1.materialization=False
cte2.materialization=True
cte3.materialization=False
cte4.materialization=True
# ^ union: recursive_cte = (cte1 ^ cte2)
# Recursive CTE wit UINION only supported by: SQLite3, MySQL, Postgres.
recursive_cte = (cte1 + cte2)
recursive_cte.alias = "Hierarchy" # you have to set alias for Recursive CTE
# columnsAliases used for Oracle only but they are accepted syntax if you didn't remove it from the test for SQLite3, MySQL, Postgres, and MicrosoftSQL.
recursive_cte.columnsAliases = "employee_id, manager_id, first_name"
# ee.materialization(True) # Error: materialization with recursive
# RECURSIVE: Required for MySQL and Postgres. Optional for SQLite3.
# RECURSIVE: are not used by Oracle and MicrosoftSQL.
if Record.database__.name in ['Oracle', 'MicrosoftSQL']:
with_cte = WithCTE((cte3 >> cte4 >> recursive_cte), recursive=False)
elif Record.database__.name in ['Postgres']:
"""
CYCLE Detection (PostgreSQL 14+): CYCLE employee_id SET is_cycle USING path
SEARCH Clause (PostgreSQL 14+): SEARCH DEPTH FIRST BY employee_id SET ordercol
"""
# with_cte = WithCTE((cte3 >> cte4 >> recursive_cte), recursive=True, options='CYCLE employee_id SET is_cycle USING path')
with_cte = WithCTE((cte3 >> cte4 >> recursive_cte), recursive=True, options='SEARCH DEPTH FIRST BY employee_id SET ordercol')
else:# ['SQLite3', 'MySQL']
with_cte = WithCTE((cte3 >> cte4 >> recursive_cte), recursive=True)
sql_query = f"{with_cte.value} SELECT * FROM Hierarchy" # build on top of generated WITH CTE
# print(sql_query)
print("Raw SQL:")
# Run SELECT WITH CTE as raw/plain SQL
rec = Record(statement=sql_query, parameters=with_cte.parameters, operation=Database.select)
for r in rec:
print(r.data)
print("Cartonnage:")
# SQLite3/Postgres:
WITH RECURSIVE
ExecutivesDepartment AS (
SELECT /*+ INLINE */ ExecutivesDepartment.* FROM Departments ExecutivesDepartment
WHERE ExecutivesDepartment.department_name = ?
),
AdministrationJobs AS (
SELECT /*+ MATERIALIZE */ AdministrationJobs.* FROM Jobs AdministrationJobs
WHERE AdministrationJobs.job_title LIKE ?
),
Hierarchy (employee_id, manager_id, first_name) AS (
SELECT /*+ INLINE */ employee_id, manager_id, first_name FROM Employees P
WHERE P.manager_id IS NULL
UNION ALL
SELECT /*+ MATERIALIZE */ C.employee_id, C.manager_id, C.first_name
FROM Employees C
INNER JOIN Hierarchy Hierarchy ON C.manager_id = Hierarchy.employee_id
WHERE C.first_name = ?
)
['Executive', 'Administration%', 'Neena']
if(Record.database__.name not in ['Oracle']):
emp = (
Employees()
.with_cte(with_cte)
.where(Employees.employee_id.in_subquery(hierarchy, selected='employee_id'))
.set(salary = 2000)
.update()
)
# sqlite3 returns -1 for complex operations not the real affected rows count
if(Record.database__.name in ['SQLite3']):
assert emp.rowsCount() == -1, emp.rowsCount()
else:
assert emp.rowsCount() == 2, emp.rowsCount()
# print(f"{'-'*80}")
# print(emp.query__.statement)
# print(emp.query__.parameters)
WITH RECURSIVE ... AS (...)
UPDATE Employees SET salary=?
WHERE Employees.employee_id IN (
SELECT employee_id FROM Hierarchy Hierarchy
WHERE 1=1
)
['Executive', 'Administration%', 'Neena', 2000]
emp = (
Employees()
.with_cte(with_cte)
.join(Hierarchy, (Employees.employee_id == Hierarchy.employee_id))
.join(ExecutivesDepartment, (Employees.department_id == ExecutivesDepartment.department_id))
.join(AdministrationJobs, (Employees.job_id == AdministrationJobs.job_id))
)
if Record.database__.name in ['MicrosoftSQL']:
# MSSQL MAXRECURSION Option: OPTION (MAXRECURSION 100)
emp.select(option='OPTION (MAXRECURSION 100)')
else:
emp.select()
for r in emp:
print(r.data)
# print(f"{'-'*80}")
# print(emp.query__.statement)
# print(emp.query__.parameters)
WITH RECURSIVE ... AS (...)
SELECT * FROM Employees Employees
INNER JOIN Hierarchy Hierarchy ON Employees.employee_id = Hierarchy.employee_id
INNER JOIN Departments ExecutivesDepartment ON Employees.department_id = ExecutivesDepartment.department_id
INNER JOIN Jobs AdministrationJobs ON Employees.job_id = AdministrationJobs.job_id
WHERE 1=1
# MicrosoftSQL:
... OPTION (MAXRECURSION 100)
['Executive', 'Administration%', 'Neena']
if(Record.database__.name not in ['Oracle']):
emp = (
Employees()
.with_cte(with_cte)
.where(Employees.employee_id.in_subquery(hierarchy, selected='employee_id'))
.delete()
)
# sqlite3 returns -1 for complex operations not the real affected rows count
if(Record.database__.name in ['SQLite3']):
assert emp.rowsCount() == -1, emp.rowsCount()
else:
assert emp.rowsCount() == 2, emp.rowsCount()
# print(f"{'-'*80}")
# print(emp.query__.statement)
# print(emp.query__.parameters)
# print("After delete - checking if in transaction:")
# print(f"autocommit: {emp.database__._Database__connection.autocommit if hasattr(emp.database__._Database__connection, 'autocommit') else 'N/A'}")
Record.database__.rollback() # Force rollback
WITH RECURSIVE ... AS (...)
DELETE FROM Employees
WHERE Employees.employee_id IN (
SELECT employee_id FROM Hierarchy Hierarchy
WHERE 1=1
)
['Executive', 'Administration%', 'Neena']
employees = Recordset()
emp1 = (
Employees()
.value(employee_id = 100)
.select()
).set(last_name='Ahmed')
emp2 = (
Employees()
.value(employee_id = 101)
.select()
).set(last_name='kamal')
employees.add(emp1, emp2)
if not (Record.database__.name == "Oracle"):
# add with_cte and filters to Recordset through it's first Record instance
emp1.with_cte(with_cte).where(Employees.employee_id.in_subquery(hierarchy, selected='employee_id'))
# use onColumns if you are not sure all columns are null free
employees.update(onColumns=["employee_id"])
# Insert, Update, and Delete with Recursive CTE is not tracked on SQLite3 and MicrosoftSQL
if(Record.database__.name in ["SQLite3", "MicrosoftSQL"]):
assert employees.rowsCount() == -1, employees.rowsCount
else:
assert employees.rowsCount() == 2, employees.rowsCount
WITH RECURSIVE ... AS (...)
UPDATE Employees SET last_name=?
WHERE Employees.employee_id = ? AND Employees.employee_id IN (
SELECT employee_id FROM Hierarchy Hierarchy
WHERE 1=1
)
[('Executive', 'Administration%', 'Neena', 'Ahmed', 100), ('Executive', 'Administration%', 'Neena', 'kamal', 101)]
# This will delete only 100, But 101 will not be deleted ? why because database will evaluate CTE after each deletion !
# after deleteing 100 which is the only parent with manager_id=null will be no parent qualified with manager_id=null
# so when no parent then no childs ! so 101 will not available with the second iteration to be deleted.
employees.delete(onColumns=["employee_id"])
if(Record.database__.name in ["SQLite3", "MicrosoftSQL"]):
assert employees.rowsCount() == -1, employees.rowsCount
else:
assert employees.rowsCount() == 1, employees.rowsCount
availableEmployeesAfterDeletion = (
Employees()
.where(Employees.employee_id.in_([100,101]))
.select(selected="employee_id")
)
availableEmployees = [{'employee_id': 101}]
assert availableEmployeesAfterDeletion.recordset.data == availableEmployees, availableEmployeesAfterDeletion.recordset.data
WITH RECURSIVE ... AS (...)
DELETE FROM Employees
WHERE Employees.employee_id = ? AND Employees.employee_id IN (
SELECT employee_id FROM Hierarchy Hierarchy
WHERE 1=1
)
# Note: After first DELETE, CTE is re-evaluated.
# If hierarchy root is deleted, child records no longer match.
[('Executive', 'Administration%', 'Neena', 100), ('Executive', 'Administration%', 'Neena', 101)]
# now I will delete 101 to test Recordset insertion WithCTE
# use onColumns if you are not sure all columns are null free
availableEmployeesAfterDeletion.delete()
# Record.database__._Database__cursor.execute("SELECT employee_id FROM Employees WHERE employee_id IN (100, 101)")
# print("RAW CHECK:", Record.database__._Database__cursor.fetchall())
if (Record.database__.name not in ["Oracle", "MySQL"]):
employees.insert()
if(Record.database__.name in ["SQLite3", "MicrosoftSQL"]):
assert employees.rowsCount() == -1, employees.rowsCount
else:
assert employees.rowsCount() == 2, employees.rowsCount
Record.database__.rollback() # Force rollback
WITH RECURSIVE ... AS (...)
INSERT INTO Employees (employee_id, first_name, last_name, email, phone_number, hire_date, job_id, salary, department_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
# Note: INSERT with CTE not supported by Oracle and MySQL
[('Executive', 'Administration%', 'Neena', 100, 'Steven', 'Ahmed', ...), ('Executive', 'Administration%', 'Neena', 101, 'Neena', 'kamal', ...)]