1"""
2Relational Model Fundamentals
3
4Demonstrates core concepts of the relational model:
5- Relations (tables) with domains, tuples (rows), attributes (columns)
6- Keys: superkey, candidate key, primary key, foreign key
7- Integrity constraints: entity integrity, referential integrity
8- NULL handling and 3-valued logic (TRUE, FALSE, UNKNOWN)
9
10Theory:
11- A relation is a subset of the Cartesian product of domains
12- Entity integrity: Primary key cannot be NULL
13- Referential integrity: Foreign key must reference existing primary key or be NULL
14- 3-valued logic: NULL comparisons yield UNKNOWN, affecting WHERE clause results
15"""
16
17import sqlite3
18from typing import List, Tuple
19
20
21def create_database() -> sqlite3.Connection:
22 """Create an in-memory database with proper constraints."""
23 conn = sqlite3.connect(':memory:')
24 cursor = conn.cursor()
25
26 # Domain constraints enforced through data types and CHECK constraints
27 cursor.execute('''
28 CREATE TABLE Departments (
29 dept_id INTEGER PRIMARY KEY, -- Entity integrity: NOT NULL implicit
30 dept_name TEXT NOT NULL UNIQUE, -- Candidate key
31 budget REAL CHECK(budget >= 0), -- Domain constraint
32 manager_id INTEGER -- Can be NULL (no manager yet)
33 )
34 ''')
35
36 cursor.execute('''
37 CREATE TABLE Employees (
38 emp_id INTEGER PRIMARY KEY, -- Entity integrity
39 emp_name TEXT NOT NULL,
40 salary REAL CHECK(salary > 0), -- Domain constraint
41 dept_id INTEGER, -- Foreign key (can be NULL for unassigned employees)
42 hire_date TEXT, -- ISO format: YYYY-MM-DD
43 FOREIGN KEY (dept_id) REFERENCES Departments(dept_id)
44 ON DELETE SET NULL -- Referential integrity action
45 ON UPDATE CASCADE
46 )
47 ''')
48
49 # Self-referencing foreign key (employee reports to another employee)
50 cursor.execute('''
51 CREATE TABLE ReportsTo (
52 emp_id INTEGER PRIMARY KEY,
53 manager_id INTEGER,
54 FOREIGN KEY (emp_id) REFERENCES Employees(emp_id)
55 ON DELETE CASCADE,
56 FOREIGN KEY (manager_id) REFERENCES Employees(emp_id)
57 ON DELETE SET NULL
58 )
59 ''')
60
61 conn.commit()
62 return conn
63
64
65def demonstrate_keys(conn: sqlite3.Connection):
66 """Demonstrate different types of keys."""
67 print("=" * 60)
68 print("KEYS AND CONSTRAINTS")
69 print("=" * 60)
70
71 cursor = conn.cursor()
72
73 # Insert departments
74 cursor.execute("INSERT INTO Departments VALUES (1, 'Engineering', 500000, NULL)")
75 cursor.execute("INSERT INTO Departments VALUES (2, 'Sales', 300000, NULL)")
76 cursor.execute("INSERT INTO Departments VALUES (3, 'HR', 150000, NULL)")
77
78 # Try to violate primary key constraint
79 print("\n1. PRIMARY KEY (Entity Integrity)")
80 print("-" * 60)
81 try:
82 cursor.execute("INSERT INTO Departments VALUES (1, 'Marketing', 200000, NULL)")
83 print("ERROR: Should have failed (duplicate primary key)")
84 except sqlite3.IntegrityError as e:
85 print(f"✓ Primary key violation caught: {e}")
86
87 # Try to insert NULL primary key
88 try:
89 cursor.execute("INSERT INTO Departments VALUES (NULL, 'Marketing', 200000, NULL)")
90 print("ERROR: Should have failed (NULL primary key)")
91 except sqlite3.IntegrityError as e:
92 print(f"✓ NULL primary key violation caught: {e}")
93
94 # Candidate key (UNIQUE constraint)
95 print("\n2. CANDIDATE KEY (dept_name is UNIQUE)")
96 print("-" * 60)
97 try:
98 cursor.execute("INSERT INTO Departments VALUES (4, 'Engineering', 100000, NULL)")
99 print("ERROR: Should have failed (duplicate candidate key)")
100 except sqlite3.IntegrityError as e:
101 print(f"✓ Candidate key violation caught: {e}")
102
103 conn.commit()
104
105
106def demonstrate_referential_integrity(conn: sqlite3.Connection):
107 """Demonstrate referential integrity constraints."""
108 print("\n" + "=" * 60)
109 print("REFERENTIAL INTEGRITY")
110 print("=" * 60)
111
112 cursor = conn.cursor()
113
114 # Insert employees
115 cursor.execute("INSERT INTO Employees VALUES (101, 'Alice', 90000, 1, '2020-01-15')")
116 cursor.execute("INSERT INTO Employees VALUES (102, 'Bob', 85000, 1, '2020-03-10')")
117 cursor.execute("INSERT INTO Employees VALUES (103, 'Charlie', 70000, 2, '2021-06-01')")
118 cursor.execute("INSERT INTO Employees VALUES (104, 'Diana', 65000, NULL, '2022-02-14')") # NULL dept_id allowed
119
120 # Try to violate foreign key constraint
121 print("\n1. FOREIGN KEY Constraint")
122 print("-" * 60)
123 try:
124 cursor.execute("INSERT INTO Employees VALUES (105, 'Eve', 75000, 999, '2022-05-20')")
125 print("ERROR: Should have failed (invalid foreign key)")
126 except sqlite3.IntegrityError as e:
127 print(f"✓ Foreign key violation caught: {e}")
128
129 # Test ON DELETE SET NULL
130 print("\n2. ON DELETE SET NULL")
131 print("-" * 60)
132 print("Before delete:")
133 cursor.execute("SELECT emp_id, emp_name, dept_id FROM Employees WHERE dept_id = 2")
134 print(f" Employee in dept 2: {cursor.fetchall()}")
135
136 cursor.execute("DELETE FROM Departments WHERE dept_id = 2")
137 cursor.execute("SELECT emp_id, emp_name, dept_id FROM Employees WHERE emp_id = 103")
138 print(f"After deleting dept 2:")
139 print(f" Employee 103 dept_id: {cursor.fetchone()}")
140
141 # Test ON UPDATE CASCADE
142 print("\n3. ON UPDATE CASCADE")
143 print("-" * 60)
144 cursor.execute("UPDATE Departments SET dept_id = 10 WHERE dept_id = 1")
145 cursor.execute("SELECT emp_id, emp_name, dept_id FROM Employees WHERE emp_id IN (101, 102)")
146 print("After updating dept_id 1 → 10:")
147 for row in cursor.fetchall():
148 print(f" Emp {row[0]} ({row[1]}): dept_id = {row[2]}")
149
150 conn.commit()
151
152
153def demonstrate_null_logic(conn: sqlite3.Connection):
154 """Demonstrate 3-valued logic with NULL."""
155 print("\n" + "=" * 60)
156 print("NULL HANDLING AND 3-VALUED LOGIC")
157 print("=" * 60)
158
159 cursor = conn.cursor()
160
161 # Show employees with their dept_id
162 print("\n1. Current employees:")
163 print("-" * 60)
164 cursor.execute("SELECT emp_id, emp_name, dept_id FROM Employees")
165 for row in cursor.fetchall():
166 print(f" Emp {row[0]} ({row[1]}): dept_id = {row[2]}")
167
168 # NULL comparisons
169 print("\n2. NULL Comparisons (3-valued logic)")
170 print("-" * 60)
171
172 # This returns only rows where dept_id is not NULL and equals 10
173 cursor.execute("SELECT COUNT(*) FROM Employees WHERE dept_id = 10")
174 print(f" dept_id = 10: {cursor.fetchone()[0]} rows")
175
176 # This returns only rows where dept_id is not NULL and not equals 10
177 cursor.execute("SELECT COUNT(*) FROM Employees WHERE dept_id != 10")
178 print(f" dept_id != 10: {cursor.fetchone()[0]} rows")
179
180 # NULL is neither equal nor not equal to anything
181 cursor.execute("SELECT COUNT(*) FROM Employees WHERE dept_id IS NULL")
182 null_count = cursor.fetchone()[0]
183 print(f" dept_id IS NULL: {null_count} rows")
184
185 cursor.execute("SELECT COUNT(*) FROM Employees WHERE dept_id IS NOT NULL")
186 not_null_count = cursor.fetchone()[0]
187 print(f" dept_id IS NOT NULL: {not_null_count} rows")
188
189 cursor.execute("SELECT COUNT(*) FROM Employees")
190 total = cursor.fetchone()[0]
191 print(f" Total rows: {total}")
192 print(f" ✓ NULL + NOT NULL = Total: {null_count + not_null_count} = {total}")
193
194 # COALESCE for NULL handling
195 print("\n3. COALESCE for NULL handling")
196 print("-" * 60)
197 cursor.execute("""
198 SELECT emp_id, emp_name,
199 COALESCE(dept_id, -1) as dept_id_or_default
200 FROM Employees
201 """)
202 for row in cursor.fetchall():
203 print(f" Emp {row[0]} ({row[1]}): dept_id = {row[2]} (-1 means unassigned)")
204
205
206def demonstrate_domains(conn: sqlite3.Connection):
207 """Demonstrate domain constraints."""
208 print("\n" + "=" * 60)
209 print("DOMAIN CONSTRAINTS")
210 print("=" * 60)
211
212 cursor = conn.cursor()
213
214 print("\n1. CHECK Constraint (salary > 0)")
215 print("-" * 60)
216 try:
217 cursor.execute("INSERT INTO Employees VALUES (106, 'Frank', -1000, 10, '2023-01-01')")
218 print("ERROR: Should have failed (negative salary)")
219 except sqlite3.IntegrityError as e:
220 print(f"✓ Domain constraint violation: {e}")
221
222 print("\n2. NOT NULL Constraint")
223 print("-" * 60)
224 try:
225 cursor.execute("INSERT INTO Employees VALUES (106, NULL, 75000, 10, '2023-01-01')")
226 print("ERROR: Should have failed (NULL name)")
227 except sqlite3.IntegrityError as e:
228 print(f"✓ NOT NULL constraint violation: {e}")
229
230 print("\n3. Type Affinity (SQLite feature)")
231 print("-" * 60)
232 # SQLite has type affinity, not strict typing
233 cursor.execute("INSERT INTO Employees VALUES (106, 'Frank', '75000', 10, '2023-01-01')")
234 cursor.execute("SELECT emp_id, salary, typeof(salary) FROM Employees WHERE emp_id = 106")
235 row = cursor.fetchone()
236 print(f"Inserted salary as string '75000':")
237 print(f" Retrieved value: {row[1]}, type: {row[2]}")
238 print(" (SQLite converts to REAL due to column affinity)")
239
240 conn.commit()
241
242
243if __name__ == "__main__":
244 print("""
245╔══════════════════════════════════════════════════════════════╗
246║ RELATIONAL MODEL FUNDAMENTALS ║
247║ Domains, Keys, Integrity Constraints, NULL Logic ║
248╚══════════════════════════════════════════════════════════════╝
249""")
250
251 conn = create_database()
252
253 try:
254 demonstrate_keys(conn)
255 demonstrate_referential_integrity(conn)
256 demonstrate_null_logic(conn)
257 demonstrate_domains(conn)
258
259 print("\n" + "=" * 60)
260 print("SUMMARY")
261 print("=" * 60)
262 print("✓ Relational model enforces data integrity through:")
263 print(" - Entity integrity (primary keys)")
264 print(" - Referential integrity (foreign keys)")
265 print(" - Domain constraints (types, CHECK, NOT NULL)")
266 print(" - 3-valued logic for NULL handling")
267 print("=" * 60)
268
269 finally:
270 conn.close()