Learn how to connect Python applications to databases using SQLite and SQL, perform CRUD operations, and build data-driven applications.
Database integration is crucial for building data-driven applications. This lesson covers how to work with SQLite databases in Python, perform SQL operations, manage data persistence, and build robust database-backed applications using Python's built-in sqlite3 module.
Relational databases store data in tables with rows and columns, connected through relationships. Understanding these fundamentals is essential for effective database integration.
Database Concepts:
1-- Basic table structure
2CREATE TABLE students (
3 student_id INTEGER PRIMARY KEY,
4 first_name TEXT NOT NULL,
5 last_name TEXT NOT NULL,
6 email TEXT UNIQUE NOT NULL,
7 gpa REAL CHECK (gpa >= 0.0 AND gpa <= 4.0)
8);
9
10-- Relationships with foreign keys
11CREATE TABLE enrollments (
12 enrollment_id INTEGER PRIMARY KEY,
13 student_id INTEGER,
14 course_id INTEGER,
15 FOREIGN KEY (student_id) REFERENCES students (student_id),
16 FOREIGN KEY (course_id) REFERENCES courses (course_id)
17);SQL Command Types:
Understanding primary keys, foreign keys, and basic joins is essential for effective database design and maintaining data integrity.
SQLite is a lightweight, file-based database perfect for development and small applications. Python's sqlite3 module provides a simple interface to SQLite databases.
Basic SQLite Operations:
1import sqlite3
2
3# Establish connection
4connection = sqlite3.connect('database.db')
5cursor = connection.cursor()
6
7# Create table
8cursor.execute('''
9 CREATE TABLE IF NOT EXISTS users (
10 id INTEGER PRIMARY KEY AUTOINCREMENT,
11 name TEXT NOT NULL,
12 email TEXT UNIQUE NOT NULL
13 )
14''')
15
16# Insert data (parameterized query)
17cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)",
18 ("John Doe", "[email protected]"))
19
20# Query data
21cursor.execute("SELECT * FROM users WHERE name LIKE ?", ("%John%",))
22results = cursor.fetchall()
23
24# Commit changes and close
25connection.commit()
26connection.close()Key Concepts:
Always use parameterized queries and properly close connections to prevent data corruption.
CRUD operations form the foundation of data manipulation. Each operation requires careful implementation with proper error handling and data validation.
Complete CRUD Implementation:
1class StudentDatabase:
2 def __init__(self, db_path):
3 self.db_path = db_path
4 self.setup_database()
5
6 def create_student(self, name, email, gpa):
7 """CREATE - Insert new student"""
8 try:
9 with sqlite3.connect(self.db_path) as conn:
10 cursor = conn.cursor()
11 cursor.execute(
12 "INSERT INTO students (name, email, gpa) VALUES (?, ?, ?)",
13 (name, email, gpa)
14 )
15 return cursor.lastrowid
16 except sqlite3.IntegrityError:
17 raise ValueError("Email already exists")
18
19 def read_student(self, student_id):
20 """READ - Retrieve student by ID"""
21 with sqlite3.connect(self.db_path) as conn:
22 conn.row_factory = sqlite3.Row
23 cursor = conn.cursor()
24 cursor.execute("SELECT * FROM students WHERE id = ?", (student_id,))
25 return cursor.fetchone()
26
27 def update_student(self, student_id, name=None, email=None, gpa=None):
28 """UPDATE - Modify existing student"""
29 updates = []
30 params = []
31
32 if name:
33 updates.append("name = ?")
34 params.append(name)
35 if email:
36 updates.append("email = ?")
37 params.append(email)
38 if gpa is not None:
39 updates.append("gpa = ?")
40 params.append(gpa)
41
42 if updates:
43 params.append(student_id)
44 query = f"UPDATE students SET {', '.join(updates)} WHERE id = ?"
45
46 with sqlite3.connect(self.db_path) as conn:
47 cursor = conn.cursor()
48 cursor.execute(query, params)
49 return cursor.rowcount > 0
50 return False
51
52 def delete_student(self, student_id):
53 """DELETE - Remove student"""
54 with sqlite3.connect(self.db_path) as conn:
55 cursor = conn.cursor()
56 cursor.execute("DELETE FROM students WHERE id = ?", (student_id,))
57 return cursor.rowcount > 0
58
59 def search_students(self, search_term):
60 """Advanced READ - Search with filtering"""
61 with sqlite3.connect(self.db_path) as conn:
62 conn.row_factory = sqlite3.Row
63 cursor = conn.cursor()
64 cursor.execute(
65 "SELECT * FROM students WHERE name LIKE ? OR email LIKE ? ORDER BY name",
66 (f"%{search_term}%", f"%{search_term}%")
67 )
68 return cursor.fetchall()Each CRUD operation includes proper error handling, data validation, and safe parameter binding to prevent SQL injection attacks.
Transactions ensure data integrity by grouping related operations that either all succeed or all fail. Implementing proper transaction management and security practices is crucial for robust database applications.
Transaction Management:
1class TransactionManager:
2 def __init__(self, db_path):
3 self.db_path = db_path
4
5 def transfer_student_credits(self, from_student_id, to_student_id, credits):
6 """Example of transaction with rollback capability"""
7 try:
8 with sqlite3.connect(self.db_path) as conn:
9 # Start transaction (automatic with 'with' statement)
10 cursor = conn.cursor()
11
12 # Check if source student has enough credits
13 cursor.execute("SELECT credits FROM students WHERE id = ?", (from_student_id,))
14 source_credits = cursor.fetchone()[0]
15
16 if source_credits < credits:
17 raise ValueError("Insufficient credits")
18
19 # Deduct from source
20 cursor.execute(
21 "UPDATE students SET credits = credits - ? WHERE id = ?",
22 (credits, from_student_id)
23 )
24
25 # Add to destination
26 cursor.execute(
27 "UPDATE students SET credits = credits + ? WHERE id = ?",
28 (credits, to_student_id)
29 )
30
31 # Transaction commits automatically if no exception
32 print(f"Successfully transferred {credits} credits")
33
34 except Exception as e:
35 # Transaction rolls back automatically on exception
36 print(f"Transfer failed: {e}")
37 raise
38
39# Advanced error handling and connection management
40class DatabaseManager:
41 def __init__(self, db_path):
42 self.db_path = db_path
43 self.connection_pool = []
44
45 def get_connection(self):
46 """Get database connection with error handling"""
47 try:
48 conn = sqlite3.connect(self.db_path, timeout=10.0)
49 conn.row_factory = sqlite3.Row
50 conn.execute("PRAGMA foreign_keys = ON") # Enable foreign key constraints
51 return conn
52 except sqlite3.Error as e:
53 print(f"Database connection error: {e}")
54 raise
55
56 def execute_with_retry(self, query, params=(), max_retries=3):
57 """Execute query with automatic retry for transient errors"""
58 for attempt in range(max_retries):
59 try:
60 with self.get_connection() as conn:
61 cursor = conn.cursor()
62 cursor.execute(query, params)
63 return cursor.fetchall()
64 except sqlite3.OperationalError as e:
65 if "database is locked" in str(e) and attempt < max_retries - 1:
66 time.sleep(0.1 * (2 ** attempt)) # Exponential backoff
67 continue
68 raiseBest Practices:
with statements) for automatic transaction managementFollowing these practices ensures your database applications are robust, secure, and maintainable.
Build a comprehensive student management system with database operations, demonstrating real-world database integration patterns
1import sqlite3
2import datetime
3from typing import List, Dict, Optional, Tuple
4from dataclasses import dataclass
5import json
6
7@dataclass
8class Student:
9 """Student data model"""
10 student_id: Optional[int] = None
11 first_name: str = ""
12 last_name: str = ""
13 email: str = ""
14 date_of_birth: str = ""
15 enrollment_date: str = ""
16 major: str = ""
17 gpa: float = 0.0
18
19 def to_dict(self) -> Dict:
20 """Convert student to dictionary"""
21 return {
22 'student_id': self.student_id,
23 'first_name': self.first_name,
24 'last_name': self.last_name,
25 'email': self.email,
26 'date_of_birth': self.date_of_birth,
27 'enrollment_date': self.enrollment_date,
28 'major': self.major,
29 'gpa': self.gpa
30 }
31
32@dataclass
33class Course:
34 """Course data model"""
35 course_id: Optional[int] = None
36 course_code: str = ""
37 course_name: str = ""
38 credits: int = 0
39 instructor: str = ""
40 semester: str = ""
41
42class DatabaseManager:
43 """Database connection and management class"""
44
45 def __init__(self, db_path: str = "student_management.db"):
46 self.db_path = db_path
47 self.connection = None
48 self.setup_database()
49
50 def connect(self):
51 """Establish database connection"""
52 try:
53 self.connection = sqlite3.connect(self.db_path)
54 self.connection.row_factory = sqlite3.Row # Enable dict-like access
55 return self.connection
56 except sqlite3.Error as e:
57 print(f"Database connection error: {e}")
58 raise
59
60 def disconnect(self):
61 """Close database connection"""
62 if self.connection:
63 self.connection.close()
64 self.connection = None
65
66 def execute_query(self, query: str, params: Tuple = ()) -> Optional[sqlite3.Cursor]:
67 """Execute a query with parameters"""
68 try:
69 if not self.connection:
70 self.connect()
71
72 cursor = self.connection.cursor()
73 cursor.execute(query, params)
74 return cursor
75
76 except sqlite3.Error as e:
77 print(f"Query execution error: {e}")
78 self.connection.rollback()
79 raise
80
81 def setup_database(self):
82 """Create database tables if they don't exist"""
83 self.connect()
84
85 # Students table
86 students_table = """
87 CREATE TABLE IF NOT EXISTS students (
88 student_id INTEGER PRIMARY KEY AUTOINCREMENT,
89 first_name TEXT NOT NULL,
90 last_name TEXT NOT NULL,
91 email TEXT UNIQUE NOT NULL,
92 date_of_birth DATE NOT NULL,
93 enrollment_date DATE NOT NULL DEFAULT CURRENT_DATE,
94 major TEXT NOT NULL,
95 gpa REAL DEFAULT 0.0 CHECK (gpa >= 0.0 AND gpa <= 4.0)
96 )
97 """
98
99 # Courses table
100 courses_table = """
101 CREATE TABLE IF NOT EXISTS courses (
102 course_id INTEGER PRIMARY KEY AUTOINCREMENT,
103 course_code TEXT UNIQUE NOT NULL,
104 course_name TEXT NOT NULL,
105 credits INTEGER NOT NULL CHECK (credits > 0),
106 instructor TEXT NOT NULL,
107 semester TEXT NOT NULL
108 )
109 """
110
111 # Enrollments table (many-to-many relationship)
112 enrollments_table = """
113 CREATE TABLE IF NOT EXISTS enrollments (
114 enrollment_id INTEGER PRIMARY KEY AUTOINCREMENT,
115 student_id INTEGER NOT NULL,
116 course_id INTEGER NOT NULL,
117 enrollment_date DATE NOT NULL DEFAULT CURRENT_DATE,
118 grade TEXT DEFAULT 'IP', -- IP = In Progress
119 FOREIGN KEY (student_id) REFERENCES students (student_id)
120 ON DELETE CASCADE,
121 FOREIGN KEY (course_id) REFERENCES courses (course_id)
122 ON DELETE CASCADE,
123 UNIQUE (student_id, course_id)
124 )
125 """
126
127 # Grades table
128 grades_table = """
129 CREATE TABLE IF NOT EXISTS grades (
130 grade_id INTEGER PRIMARY KEY AUTOINCREMENT,
131 enrollment_id INTEGER NOT NULL,
132 assignment_name TEXT NOT NULL,
133 points_earned REAL NOT NULL CHECK (points_earned >= 0),
134 points_possible REAL NOT NULL CHECK (points_possible > 0),
135 grade_date DATE NOT NULL DEFAULT CURRENT_DATE,
136 FOREIGN KEY (enrollment_id) REFERENCES enrollments (enrollment_id)
137 ON DELETE CASCADE
138 )
139 """
140
141 try:
142 self.execute_query(students_table)
143 self.execute_query(courses_table)
144 self.execute_query(enrollments_table)
145 self.execute_query(grades_table)
146 self.connection.commit()
147 print("Database tables created successfully")
148
149 # Create indexes for performance
150 self.create_indexes()
151
152 except sqlite3.Error as e:
153 print(f"Database setup error: {e}")
154 self.connection.rollback()
155 raise
156
157 def create_indexes(self):
158 """Create indexes for better query performance"""
159 indexes = [
160 "CREATE INDEX IF NOT EXISTS idx_students_email ON students(email)",
161 "CREATE INDEX IF NOT EXISTS idx_students_major ON students(major)",
162 "CREATE INDEX IF NOT EXISTS idx_courses_code ON courses(course_code)",
163 "CREATE INDEX IF NOT EXISTS idx_enrollments_student ON enrollments(student_id)",
164 "CREATE INDEX IF NOT EXISTS idx_enrollments_course ON enrollments(course_id)",
165 "CREATE INDEX IF NOT EXISTS idx_grades_enrollment ON grades(enrollment_id)"
166 ]
167
168 for index_sql in indexes:
169 self.execute_query(index_sql)
170
171class StudentManager:
172 """Student management operations"""
173
174 def __init__(self, db_manager: DatabaseManager):
175 self.db = db_manager
176
177 def add_student(self, student: Student) -> int:
178 """Add a new student to the database"""
179 query = """
180 INSERT INTO students (first_name, last_name, email, date_of_birth, enrollment_date, major, gpa)
181 VALUES (?, ?, ?, ?, ?, ?, ?)
182 """
183
184 try:
185 # Validate email format (simple validation)
186 if '@' not in student.email:
187 raise ValueError("Invalid email format")
188
189 # Set default enrollment date if not provided
190 if not student.enrollment_date:
191 student.enrollment_date = datetime.date.today().strftime('%Y-%m-%d')
192
193 cursor = self.db.execute_query(query, (
194 student.first_name,
195 student.last_name,
196 student.email,
197 student.date_of_birth,
198 student.enrollment_date,
199 student.major,
200 student.gpa
201 ))
202
203 self.db.connection.commit()
204 return cursor.lastrowid
205
206 except sqlite3.IntegrityError:
207 raise ValueError(f"Student with email {student.email} already exists")
208
209 def get_student(self, student_id: int) -> Optional[Student]:
210 """Retrieve a student by ID"""
211 query = "SELECT * FROM students WHERE student_id = ?"
212
213 cursor = self.db.execute_query(query, (student_id,))
214 row = cursor.fetchone()
215
216 if row:
217 return Student(
218 student_id=row['student_id'],
219 first_name=row['first_name'],
220 last_name=row['last_name'],
221 email=row['email'],
222 date_of_birth=row['date_of_birth'],
223 enrollment_date=row['enrollment_date'],
224 major=row['major'],
225 gpa=row['gpa']
226 )
227 return None
228
229 def get_students_by_major(self, major: str) -> List[Student]:
230 """Get all students in a specific major"""
231 query = "SELECT * FROM students WHERE major = ? ORDER BY last_name, first_name"
232
233 cursor = self.db.execute_query(query, (major,))
234 students = []
235
236 for row in cursor.fetchall():
237 student = Student(
238 student_id=row['student_id'],
239 first_name=row['first_name'],
240 last_name=row['last_name'],
241 email=row['email'],
242 date_of_birth=row['date_of_birth'],
243 enrollment_date=row['enrollment_date'],
244 major=row['major'],
245 gpa=row['gpa']
246 )
247 students.append(student)
248
249 return students
250
251 def update_student(self, student: Student) -> bool:
252 """Update an existing student"""
253 query = """
254 UPDATE students SET
255 first_name = ?, last_name = ?, email = ?,
256 date_of_birth = ?, major = ?, gpa = ?
257 WHERE student_id = ?
258 """
259
260 try:
261 cursor = self.db.execute_query(query, (
262 student.first_name,
263 student.last_name,
264 student.email,
265 student.date_of_birth,
266 student.major,
267 student.gpa,
268 student.student_id
269 ))
270
271 self.db.connection.commit()
272 return cursor.rowcount > 0
273
274 except sqlite3.IntegrityError:
275 raise ValueError(f"Email {student.email} is already in use")
276
277 def delete_student(self, student_id: int) -> bool:
278 """Delete a student and related records"""
279 try:
280 # Start transaction
281 self.db.connection.execute("BEGIN")
282
283 # Delete related records first (due to foreign key constraints)
284 self.db.execute_query("DELETE FROM grades WHERE enrollment_id IN (SELECT enrollment_id FROM enrollments WHERE student_id = ?)", (student_id,))
285 self.db.execute_query("DELETE FROM enrollments WHERE student_id = ?", (student_id,))
286
287 # Delete student
288 cursor = self.db.execute_query("DELETE FROM students WHERE student_id = ?", (student_id,))
289
290 self.db.connection.commit()
291 return cursor.rowcount > 0
292
293 except sqlite3.Error:
294 self.db.connection.rollback()
295 raise
296
297 def search_students(self, search_term: str) -> List[Student]:
298 """Search students by name or email"""
299 query = """
300 SELECT * FROM students
301 WHERE first_name LIKE ? OR last_name LIKE ? OR email LIKE ?
302 ORDER BY last_name, first_name
303 """
304
305 search_pattern = f"%{search_term}%"
306 cursor = self.db.execute_query(query, (search_pattern, search_pattern, search_pattern))
307
308 students = []
309 for row in cursor.fetchall():
310 student = Student(
311 student_id=row['student_id'],
312 first_name=row['first_name'],
313 last_name=row['last_name'],
314 email=row['email'],
315 date_of_birth=row['date_of_birth'],
316 enrollment_date=row['enrollment_date'],
317 major=row['major'],
318 gpa=row['gpa']
319 )
320 students.append(student)
321
322 return students
323
324 def get_student_statistics(self) -> Dict:
325 """Get comprehensive student statistics"""
326 stats = {}
327
328 # Total students
329 cursor = self.db.execute_query("SELECT COUNT(*) as total FROM students")
330 stats['total_students'] = cursor.fetchone()['total']
331
332 # Students by major
333 cursor = self.db.execute_query("SELECT major, COUNT(*) as count FROM students GROUP BY major ORDER BY count DESC")
334 stats['students_by_major'] = {row['major']: row['count'] for row in cursor.fetchall()}
335
336 # GPA statistics
337 cursor = self.db.execute_query("SELECT AVG(gpa) as avg_gpa, MIN(gpa) as min_gpa, MAX(gpa) as max_gpa FROM students")
338 gpa_stats = cursor.fetchone()
339 stats['gpa_stats'] = {
340 'average': round(gpa_stats['avg_gpa'] or 0, 2),
341 'minimum': gpa_stats['min_gpa'] or 0,
342 'maximum': gpa_stats['max_gpa'] or 0
343 }
344
345 # Recent enrollments (last 30 days)
346 cursor = self.db.execute_query(
347 "SELECT COUNT(*) as recent FROM students WHERE enrollment_date >= date('now', '-30 days')"
348 )
349 stats['recent_enrollments'] = cursor.fetchone()['recent']
350
351 return stats
352
353class StudentManagementSystem:
354 """Main application class"""
355
356 def __init__(self, db_path: str = "student_management.db"):
357 self.db_manager = DatabaseManager(db_path)
358 self.student_manager = StudentManager(self.db_manager)
359
360 def populate_sample_data(self):
361 """Add sample data for demonstration"""
362 sample_students = [
363 Student(first_name="Alice", last_name="Johnson", email="[email protected]",
364 date_of_birth="2000-05-15", major="Computer Science", gpa=3.8),
365 Student(first_name="Bob", last_name="Smith", email="[email protected]",
366 date_of_birth="1999-12-03", major="Engineering", gpa=3.6),
367 Student(first_name="Charlie", last_name="Brown", email="[email protected]",
368 date_of_birth="2001-08-22", major="Computer Science", gpa=3.9),
369 Student(first_name="Diana", last_name="Wilson", email="[email protected]",
370 date_of_birth="2000-03-10", major="Mathematics", gpa=3.7),
371 Student(first_name="Eve", last_name="Davis", email="[email protected]",
372 date_of_birth="1999-11-28", major="Physics", gpa=3.5)
373 ]
374
375 print("Adding sample students...")
376 for student in sample_students:
377 try:
378 student_id = self.student_manager.add_student(student)
379 print(f"Added student: {student.first_name} {student.last_name} (ID: {student_id})")
380 except ValueError as e:
381 print(f"Skipped {student.first_name} {student.last_name}: {e}")
382
383 def demonstrate_operations(self):
384 """Demonstrate various database operations"""
385 print("\n=== Student Management System Demo ===")
386
387 # Add sample data
388 self.populate_sample_data()
389
390 # Display statistics
391 print("\n=== Student Statistics ===")
392 stats = self.student_manager.get_student_statistics()
393 print(f"Total Students: {stats['total_students']}")
394 print(f"Average GPA: {stats['gpa_stats']['average']}")
395 print(f"Recent Enrollments (30 days): {stats['recent_enrollments']}")
396
397 print("\nStudents by Major:")
398 for major, count in stats['students_by_major'].items():
399 print(f" {major}: {count}")
400
401 # Search demonstration
402 print("\n=== Search Results for 'alice' ===")
403 search_results = self.student_manager.search_students("alice")
404 for student in search_results:
405 print(f"Found: {student.first_name} {student.last_name} - {student.email}")
406
407 # Get students by major
408 print("\n=== Computer Science Students ===")
409 cs_students = self.student_manager.get_students_by_major("Computer Science")
410 for student in cs_students:
411 print(f"{student.first_name} {student.last_name} - GPA: {student.gpa}")
412
413 # Update student example
414 if cs_students:
415 student = cs_students[0]
416 original_gpa = student.gpa
417 student.gpa = 4.0
418 print(f"\n=== Updating {student.first_name} {student.last_name} GPA ===")
419 print(f"GPA changed from {original_gpa} to {student.gpa}")
420 self.student_manager.update_student(student)
421
422 print("\n=== Database Operations Complete ===")
423
424 def close(self):
425 """Clean up resources"""
426 self.db_manager.disconnect()
427
428# Example usage demonstrating comprehensive database integration
429def main():
430 """Main function demonstrating the student management system"""
431 system = None
432 try:
433 # Initialize system
434 system = StudentManagementSystem()
435
436 # Run demonstrations
437 system.demonstrate_operations()
438
439 print("\n=== Database Integration Features Demonstrated ===")
440 print("- Database table creation and schema design")
441 print("- CRUD operations with proper error handling")
442 print("- Complex queries with joins and aggregations")
443 print("- Transaction management and data integrity")
444 print("- Search functionality and data filtering")
445 print("- Statistical analysis and reporting")
446 print("- Object-relational mapping patterns")
447 print("- Database performance optimization")
448
449 except Exception as e:
450 print(f"System error: {e}")
451
452 finally:
453 if system:
454 system.close()
455
456if __name__ == "__main__":
457 main()Your progress is saved in this browser, on this device only.