Database Programming using Python

Python DB-API 2.0 (PEP 249) Architecture

पाइथनमा प्रत्येक डाटाबेस (जस्तै SQLite, MySQL, PostgreSQL, Oracle) को आफ्नै अलग-अलग C-लाइब्रेरी वा प्रोटोकल हुन्छ। यदि प्रत्येक डाटाबेसका लागि पाइथन कोड पूर्ण रूपमा फरक हुन्थ्यो भने एउटा डाटाबेसबाट अर्कोमा सर्न अत्यन्त गाह्रो हुन्थ्यो। यस समस्यालाई समाधान गर्न Python Software Foundation ले PEP 249 (Python Database API Specification v2.0) जारी गर्‍यो।

PEP 249 ले सबै डाटाबेस ड्राइभरहरूका लागि एउटै साझा मापदण्ड तोकिदिएको छ:

• `connect(...)`: डाटाबेस सर्भर वा स्थानीय फाइलसँग कनेक्सन अब्जेक्ट खोल्ने फङ्सन।
• Connection Object: यसले डाटाबेस सेसन व्यवस्थापन गर्छ र cursor(), commit(), rollback(), तथा close() विधिहरू प्रदान गर्छ।
• Cursor Object: यो वास्तविक क्वेरी चलाउने (Execute) र परिणामहरूबाट पङ्क्तिहरू (Rows) तानेर ल्याउने मुख्य माध्यम हो।

+-------------------------------------------------------------------------+
| PYTHON APPLICATION LAYER                                                |
|   import sqlite3 (Built-in) / import mysql.connector (Enterprise)        |
+-------------------------------------------------------------------------+
                                    |
            1. conn = sqlite3.connect("college.db")
                                    v
+-------------------------------------------------------------------------+
| CONNECTION OBJECT (Session Management & Transaction Boundaries)         |
|   Methods: .cursor(), .commit(), .rollback(), .close()                  |
+-------------------------------------------------------------------------+
                                    |
            2. cur = conn.cursor()
                                    v
+-------------------------------------------------------------------------+
| CURSOR OBJECT (Query Execution & Result-Set Traversal)                  |
|   Methods: .execute(), .executemany(), .fetchone(), .fetchall()         |
+-------------------------------------------------------------------------+
                                    |
            3. SQL Query Sent via Socket / Local File I/O
                                    v
+-------------------------------------------------------------------------+
| DATABASE ENGINE (DBMS: SQLite Engine / MySQL Server Daemon)              |
|   Compiles SQL, checks constraints, reads B-Tree tables & returns rows  |
+-------------------------------------------------------------------------+

Establishing Database Connections & SQLite

पाइथनमा सबैभन्दा सहज र लोकप्रिय डाटाबेस SQLite (`sqlite3`) हो, किनभने यसलाई छुट्टै सर्भर इन्स्टल वा कन्फिगर गर्नु पर्दैन। यो सिधै पाइथनको स्ट्यान्डर्ड लाइब्रेरीमा उपलब्ध हुन्छ र डिस्कमा रहेको एउटै फाइलमा सबै टेबुलहरू भण्डारण गर्छ।

१. डिस्क फाइल vs In-Memory Database (`:memory:`)

• sqlite3.connect("tu_students.db"): कम्प्युटरको हार्डड्राइभमा स्थायी फाइल बनाउँछ। प्रोग्राम बन्द भए पनि डाटा सुरक्षित रहन्छ।
• sqlite3.connect(":memory:"): कम्प्युटरको RAM मा मात्र अस्थायी डाटाबेस सिर्जना गर्छ। यो युनिट टेस्टिङ र असाध्यै तीव्र गतिको कार्यका लागि प्रयोग गरिन्छ; प्रोग्राम बन्द हुँदा डाटा नष्ट हुन्छ।

२. सुरक्षित कनेक्सन र Exception Handling

import sqlite3

def establish_connection(db_file="tu_exam.db"):
    connection = None
    try:
        # १. डाटाबेससँग कनेक्सन स्थापित गर्ने
        connection = sqlite3.connect(db_file)
        print(f"[सफलता] SQLite डाटाबेस '{db_file}' सँग सम्पर्क जोडियो।")
        print(f"SQLite Version: {sqlite3.sqlite_version}")
        return connection
    except sqlite3.Error as err:
        print(f"[त्रुटि] डाटाबेस जोड्न सकिएन: {err}")
        return None

# कनेक्सन प्रयोग गर्ने र अन्तमा बन्द गर्ने
conn = establish_connection()
if conn:
    # काम सकिएपछि कनेक्सन बन्द गर्नै पर्छ
    conn.close()
    print("डाटाबेस कनेक्सन सुरक्षित रूपमा बन्द गरियो।")

DDL Queries: CREATE, ALTER & DROP Tables

DDL (Data Definition Language) ले डाटाबेसको ढाँचा वा संरचना (Schema) परिभाषित गर्दछ। यस अन्तर्गत टेबुल सिर्जना गर्ने (CREATE), स्तम्भ थप्ने वा बदल्ने (ALTER), र टेबुल मेटाउने (DROP) पर्दछन्।

import sqlite3

conn = sqlite3.connect("mca_college.db")
cursor = conn.cursor()

# १. CREATE TABLE DDL: नयाँ विद्यार्थी तालिका बनाउने
create_table_query = """
CREATE TABLE IF NOT EXISTS students (
    student_id INTEGER PRIMARY KEY AUTOINCREMENT,
    full_name TEXT NOT NULL,
    email TEXT UNIQUE,
    semester INTEGER CHECK(semester >= 1 AND semester <= 4),
    gpa REAL DEFAULT 0.0
);
"""
cursor.execute(create_table_query)
print("[DDL] 'students' तालिका सफलतापूर्वक सिर्जना गरियो।")

# २. ALTER TABLE DDL: नयाँ स्तम्भ (Column) थप्ने
alter_query = """
ALTER TABLE students ADD COLUMN phone_number TEXT;
"""
try:
    cursor.execute(alter_query)
    print("[DDL] 'phone_number' स्तम्भ थपियो।")
except sqlite3.OperationalError:
    print("[DDL] स्तम्भ पहिले नै अवस्थित छ।")

# ३. DROP TABLE DDL (अनावश्यक टेबुल हटाउन):
# cursor.execute("DROP TABLE IF EXISTS temp_records;")

conn.close()

DML Queries & SQL Injection Prevention

DML (Data Manipulation Language) ले टेबल भित्रको डाटासँग खेल्छ। यसका मुख्य कमान्डहरू INSERT, UPDATE, र DELETE हुन्।

The Danger of String Formatting: SQL Injection Attack

यदि हामीले Python का f-strings वा string concatenation (+) प्रयोग गरेर प्रयोगकर्ताको इनपुट सिधै SQL मा मिसायौँ भने ह्याकरले सम्पूर्ण डाटाबेस चोर्न वा मेटाउन सक्छ।

VULNERABLE (F-Strings / String Concatenation):
user_email = "[email protected]' OR '1'='1"
query = f"SELECT * FROM users WHERE email = '{user_email}';"

DATABASE RECEIVES COMPILED STATEMENT:
SELECT * FROM users WHERE email = '[email protected]' OR '1'='1';
====> '1'='1' is ALWAYS TRUE! Authentication bypassed! Entire database dumped!

SAFE (Parameterized Query with Placeholders '?'):
query = "SELECT * FROM users WHERE email = ?;"
cursor.execute(query, (user_email,))

DATABASE DRIVER SEPARATES CODE FROM DATA:
1. SQL query is pre-compiled and immutable.
2. User input is treated STRICTLY as a raw string literal, neutralizing attack vectors!

Parameterized DML Operations (Single & Batch Insertion)

import sqlite3

conn = sqlite3.connect("mca_college.db")
cursor = conn.cursor()

# १. INSERT: एकल रेकर्ड सुरक्षित थप्ने (Tuple of parameters)
insert_query = "INSERT INTO students (full_name, email, semester, gpa) VALUES (?, ?, ?, ?);"
cursor.execute(insert_query, ("Prashant Adhikari", "[email protected]", 3, 3.85))

# २. executemany(): धेरै रेकर्डहरू एकैपटक ब्याच इन्सर्ट गर्ने (High Performance)
batch_students = [
    ("Smriti Rai", "[email protected]", 3, 3.92),
    ("Rabin Thapa", "[email protected]", 2, 3.45),
    ("Manish Giri", "[email protected]", 1, 3.60)
]
cursor.executemany(insert_query, batch_students)

# ३. UPDATE: रेकर्ड परिमार्जन गर्ने
update_query = "UPDATE students SET gpa = ? WHERE email = ?;"
cursor.execute(update_query, (3.90, "[email protected]"))

# ४. DELETE: रेकर्ड हटाउने
delete_query = "DELETE FROM students WHERE semester = ?;"
cursor.execute(delete_query, (1,))

# अनिवार्य: DML पछि डिस्कमा सेभ गर्न commit() कल गर्नै पर्छ!
conn.commit()
print(f"[सफलता] {cursor.rowcount} पङ्क्तिहरू प्रभावित भए र सेभ गरियो।")

conn.close()

Processing Query Result Sets: `fetchone`, `fetchmany` & `fetchall`

जब SELECT (DQL) कमान्ड चलाइन्छ, डाटाबेसले नतिजा तुरुन्तै मेमोरीमा खन्याउँदैन। बरु यसले कर्सर (Cursor) लाई पहिलो पङ्क्तिको ठीक अगाडि प्वइन्टरको रूपमा राख्छ। नतिजाहरू झिक्न DB-API ले तीनवटा विधिहरू दिन्छ:

• `cursor.fetchone()`: नतिजाबाट एउटा मात्र पङ्क्ति (Tuple) फर्काउँछ र कर्सरलाई अर्को पङ्क्तिमा सार्छ। यदि डाटा सकियो भने None फर्काउँछ।
• `cursor.fetchmany(size)`: तोकिएको संख्या (size) बराबरका पङ्क्तिहरूको लिस्ट [ (...), (...) ] फर्काउँछ (Pagination का लागि उपयुक्त)।
• `cursor.fetchall()`: बाँकी रहेका सबै पङ्क्तिहरू एकैचोटि लिस्टको रूपमा मेमोरीमा लोड गर्छ।

QUERY: cursor.execute("SELECT id, name, gpa FROM students ORDER BY id;")

[CURSOR POINTER]  ----> Before Row 1
-----------------------------------------------------------------------
Row 1: (1, "Prashant", 3.90)  <=== cursor.fetchone() returns Row 1
                                   Pointer advances to Row 2
Row 2: (2, "Smriti", 3.92)    <=== cursor.fetchmany(2) returns [Row 2, Row 3]
Row 3: (3, "Rabin", 3.45)          Pointer advances to Row 4
-----------------------------------------------------------------------
Row 4: (4, "Aayush", 3.75)    <=== cursor.fetchall() returns all remaining:
Row 5: (5, "Deepa", 3.88)          [Row 4, Row 5]
-----------------------------------------------------------------------
[CURSOR POINTER]  ----> At END-OF-DATA (Subsequent fetchone() yields None!)

`sqlite3.Row` को प्रयोग (Accessing Columns by Name)

डिफल्ट रूपमा कर्सरले नतिजालाई इन्डेक्स टुपल (जस्तै row[0], row[1]) को रूपमा दिन्छ। तर conn.row_factory = sqlite3.Row सेट गर्दा हामी डिक्सनरी जस्तै नामबाट (row["full_name"]) डाटा झिक्न सक्छौँ:

import sqlite3

conn = sqlite3.connect("mca_college.db")
# पङ्क्तिहरूलाई नामले पहुँचयोग्य बनाउन row_factory सेट गर्ने:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()

cursor.execute("SELECT student_id, full_name, gpa FROM students WHERE gpa >= ?", (3.5,))

# Memory-Efficient Iteration over Cursor
for row in cursor:
    print(f"ID: {row['student_id']} | नाम: {row['full_name']:<20} | GPA: {row['gpa']}")

conn.close()

Transaction Management: ACID, Commit & Rollback

Database Transaction भनेको एक वा धेरै SQL अपरेसनहरूको यस्तो अविभाज्य एकाइ (Single Logical Unit of Work) हो, जहाँ या त सबै अपरेसनहरू पूर्ण रूपमा सफल हुनुपर्छ, अन्यथा एउटा पनि अपरेसनको प्रभाव देखिनु हुँदैन।

ACID Properties (TU Exam Favorite Theory)

START TRANSACTION (Auto-begun by first DML in Python DB-API)
        |
        v
[STEP 1]: Deduct Rs 5,000 from Ram's Account
        |
        +---- Success? ----> Yes ----> [STEP 2]: Credit Rs 5,000 to Sita's Account
        |                                       |
        |                                       +---- Network Failure / Crash!
        v (If any step fails or error raised)   |
+------------------------------------+          v
| EXCEPTION CAUGHT!                  |   +------------------------------------+
| conn.rollback()                    |   | conn.commit()                      |
| All partial changes reversed!      |   | All steps permanently saved to     |
| Ram loses NO money!                |   | disk database file.                |
+------------------------------------+   +------------------------------------+

परीक्षामुखी Commit र Rollback कोड उदाहरण (Bank Transfer)

import sqlite3

def transfer_funds(sender_id, receiver_id, amount):
    conn = sqlite3.connect("bank.db")
    cursor = conn.cursor()

    try:
        # १. ट्रान्ज्याक्सन सुरु: पठाउनेको खाताबाट रकम घटाउने
        cursor.execute("UPDATE accounts SET balance = balance - ? WHERE acc_id = ?;", (amount, sender_id))
        
        # पठाउनेको मौज्दात ऋणमा गयो कि जाँच गर्ने (Custom Constraint Validation)
        cursor.execute("SELECT balance FROM accounts WHERE acc_id = ?;", (sender_id,))
        sender_balance = cursor.fetchone()[0]
        if sender_balance < 0:
            raise ValueError("रकम अपुग! मौज्दात भन्दा बढी पठाउन मिल्दैन।")

        # २. पाउनेको खातामा रकम थप्ने
        cursor.execute("UPDATE accounts SET balance = balance + ? WHERE acc_id = ?;", (amount, receiver_id))

        # दुवै कार्य सफलतापूर्वक सम्पन्न भएपछि मात्र स्थायी रूपमा सेभ गर्ने:
        conn.commit()
        print(f"[सफलता] रु {amount} खाता {sender_id} बाट {receiver_id} मा ट्रान्सफर भयो।")

    except Exception as error:
        # बीचमा कुनै पनि समस्या आएमा पूरै कारोबार उल्ट्याउने (Undo everything):
        conn.rollback()
        print(f"[विफलता - ROLLBACK] ट्रान्ज्याक्सन रद्द गरियो: {error}")

    finally:
        conn.close()

Full Model CRUD Project: Student Database Management

TU MCA504 परीक्षामा ८ नम्बरको प्रयोगात्मक प्रश्न आउँदा लेख्न मिल्ने पूर्ण, सफा र त्रुटिरहित पाइथन कोड:

import sqlite3

class StudentDatabaseManager:
    def __init__(self, db_name="tu_mca.db"):
        self.db_name = db_name
        self.init_database()

    def get_connection(self):
        conn = sqlite3.connect(self.db_name)
        conn.row_factory = sqlite3.Row
        return conn

    def init_database(self):
        """DDL: तालिका संरचना तयार गर्ने"""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS student_records (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    roll_no TEXT UNIQUE NOT NULL,
                    name TEXT NOT NULL,
                    subject TEXT NOT NULL,
                    marks REAL CHECK(marks >= 0 AND marks <= 100)
                );
            """)
            conn.commit()

    def add_student(self, roll, name, subject, marks):
        """DML: नयाँ विद्यार्थी थप्ने (Parameterized INSERT)"""
        try:
            with self.get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute(
                    "INSERT INTO student_records (roll_no, name, subject, marks) VALUES (?, ?, ?, ?);",
                    (roll, name, subject, marks)
                )
                conn.commit()
                print(f"[दर्ता] {name} ({roll}) सफलतापूर्वक थपियो।")
        except sqlite3.IntegrityError as e:
            print(f"[त्रुटि] रोल नम्बर {roll} पहिले नै दर्ता छ: {e}")

    def get_all_students(self):
        """DQL: सबै विद्यार्थीहरूको सूची ल्याउने (fetchall)"""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM student_records ORDER BY marks DESC;")
            return cursor.fetchall()

    def update_marks(self, roll, new_marks):
        """DML: प्राप्ताङ्क परिमार्जन गर्ने (UPDATE)"""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                "UPDATE student_records SET marks = ? WHERE roll_no = ?;",
                (new_marks, roll)
            )
            conn.commit()
            print(f"[अपडेट] रोल {roll} को नयाँ प्राप्ताङ्क: {new_marks}")

    def delete_student(self, roll):
        """DML: विद्यार्थी रेकर्ड हटाउने (DELETE)"""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM student_records WHERE roll_no = ?;", (roll,))
            conn.commit()
            print(f"[हटाइयो] रोल {roll} को विवरण मेटाइयो।")

# मुख्य परीक्षण कोड (Driver Code):
if __name__ == "__main__":
    db = StudentDatabaseManager()

    # १. इन्सर्ट गर्ने
    db.add_student("MCA-501", "Sushant Karki", "Python Programming", 88.5)
    db.add_student("MCA-502", "Pooja Shrestha", "Python Programming", 92.0)

    # २. प्रदर्शन गर्ने
    print("\n--- विद्यार्थीहरूको प्राप्ताङ्क तालिका ---")
    for s in db.get_all_students():
        print(f"Roll: {s['roll_no']:<10} | Name: {s['name']:<16} | Marks: {s['marks']}")

    # ३. अपडेट र डिलिट
    db.update_marks("MCA-501", 94.0)
    # db.delete_student("MCA-502")