Using SQLite

# Import the sqlite3 module
import sqlite3

# Step 1: Establish a connection to a database file (creates it if it doesn't exist)
conn = sqlite3.connect('example.db')

# Step 2: Create a cursor object to execute SQL commands
cur = conn.cursor()

# Step 3: Create a table named 'users'
cur.execute('''
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    username TEXT NOT NULL UNIQUE,
    email TEXT NOT NULL
)
''')

# Step 4: Insert data using parameterized queries to prevent SQL injection
users_to_add = [
    ('alice', 'alice@example.com'),
    ('bob', 'bob@example.com'),
    ('charlie', 'charlie@example.com')
]
cur.executemany('INSERT INTO users (username, email) VALUES (?, ?)', users_to_add)

# Step 5: Commit changes to save data to the database
conn.commit()

# Step 6: Query the database for all users
cur.execute('SELECT id, username, email FROM users ORDER BY username')

# Step 7: Fetch all rows and print them
rows = cur.fetchall()
for row in rows:
    print(f"User ID: {row[0]}, Username: {row[1]}, Email: {row[2]}")

# Step 8: Close the cursor and connection to free resources
cur.close()
conn.close()
    

import sqlite3

# Using context managers to handle connections and transactions safely
try:
    with sqlite3.connect('example.db') as conn:
        cur = conn.cursor()
        
        # Begin transaction implicitly with 'with' statement
        
        # Insert a new user
        cur.execute('INSERT INTO users (username, email) VALUES (?, ?)', ('diana', 'diana@example.com'))
        
        # Attempt to insert a duplicate username to trigger an error
        cur.execute('INSERT INTO users (username, email) VALUES (?, ?)', ('alice', 'alice2@example.com'))
        
        # Commit happens automatically on successful exit of 'with' block
        
except sqlite3.IntegrityError as e:
    print(f"Integrity Error caught: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")
else:
    print("Transaction committed successfully.")
    # Query to confirm new user added
    with sqlite3.connect('example.db') as conn:
        cur = conn.cursor()
        cur.execute('SELECT username, email FROM users WHERE username = ?', ('diana',))
        user = cur.fetchone()
        print(f"Added user: {user}")
    

import sqlite3

with sqlite3.connect('example.db') as conn:
    cur = conn.cursor()
    
    # Create an index to speed up queries filtering by email
    cur.execute('CREATE INDEX IF NOT EXISTS idx_email ON users(email)')
    
    # Query using the indexed column
    cur.execute('SELECT username FROM users WHERE email = ?', ('bob@example.com',))
    user = cur.fetchone()
    print(f"User with email bob@example.com: {user[0]}")
    

import sqlite3

# Create an in-memory SQLite database (volatile, lost after program ends)
with sqlite3.connect(':memory:') as conn:
    cur = conn.cursor()
    
    # Create table
    cur.execute('CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)')
    
    # Insert sample data
    cur.executemany('INSERT INTO test (value) VALUES (?)', [('foo',), ('bar',), ('baz',)])
    
    # Query data
    cur.execute('SELECT * FROM test')
    results = cur.fetchall()
    print("In-memory DB contents:", results)
    

import sqlite3
import datetime

# Adapter: Convert datetime to string before storing
def adapt_datetime(ts):
    return ts.isoformat()

# Converter: Convert string back to datetime when fetching
def convert_datetime(s):
    return datetime.datetime.fromisoformat(s.decode())

# Register adapter and converter
sqlite3.register_adapter(datetime.datetime, adapt_datetime)
sqlite3.register_converter("timestamp", convert_datetime)

# Connect with detect_types to enable converters
conn = sqlite3.connect('example.db', detect_types=sqlite3.PARSE_DECLTYPES)
cur = conn.cursor()

# Create a table with a timestamp column
cur.execute('''
CREATE TABLE IF NOT EXISTS events (
    id INTEGER PRIMARY KEY,
    event_name TEXT,
    event_time timestamp
)
''')

# Insert current time as datetime object
now = datetime.datetime.now()
cur.execute('INSERT INTO events (event_name, event_time) VALUES (?, ?)', ('Sample Event', now))

conn.commit()

# Query and fetch datetime object
cur.execute('SELECT event_name, event_time FROM events')
event = cur.fetchone()
print(f"Event: {event[0]}, Time: {event[1]} (Type: {type(event[1])})")

cur.close()
conn.close()
    

import sqlite3

# Source database connection
src_conn = sqlite3.connect('example.db')

# Destination database connection (new file)
dest_conn = sqlite3.connect('backup_example.db')

# Use backup API to copy all contents safely
with dest_conn:
    src_conn.backup(dest_conn, pages=1, progress=None)

print("Backup completed successfully!")

src_conn.close()
dest_conn.close()