185. Database Interaction

Snippet 1: Basic SQLite Interaction with sqlite3

import sqlite3

# Connect to SQLite database (creates file if it doesn't exist)
conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# Create a table
cursor.execute('CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)')
conn.commit()

# Insert data
cursor.execute('INSERT INTO users (name) VALUES (?)', ('Alice',))
conn.commit()

# Retrieve data
cursor.execute('SELECT * FROM users')
print(cursor.fetchall())

# Close connection
conn.close()

---

Snippet 2: SQLAlchemy ORM Setup

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Database setup
Base = declarative_base()
engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)
session = Session()

# Define a User model
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

# Create tables
Base.metadata.create_all(engine)

---

Snippet 3: Adding Data with SQLAlchemy ORM

# Add a new user
new_user = User(name='Bob')
session.add(new_user)
session.commit()

# Fetch all users
users = session.query(User).all()
for user in users:
    print(user.id, user.name)

---

Snippet 4: Updating Data with SQLAlchemy ORM

# Update a user's name
user = session.query(User).filter_by(name='Bob').first()
if user:
    user.name = 'Robert'
    session.commit()

# Fetch updated user
updated_user = session.query(User).filter_by(name='Robert').first()
print(updated_user.id, updated_user.name)

---

Snippet 5: Deleting Data with SQLAlchemy ORM

# Delete a user
user = session.query(User).filter_by(name='Robert').first()
if user:
    session.delete(user)
    session.commit()

# Verify deletion
remaining_users = session.query(User).all()
print([u.name for u in remaining_users])

---

Snippet 6: Using sqlite3 with Placeholders

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# Insert multiple records
data = [('Alice',), ('Charlie',), ('Dave',)]
cursor.executemany('INSERT INTO users (name) VALUES (?)', data)
conn.commit()

# Fetch users with name starting with 'A'
cursor.execute('SELECT * FROM users WHERE name LIKE ?', ('A%',))
print(cursor.fetchall())

conn.close()

---

Snippet 7: SQLAlchemy Query Filters

# Query with filters
users_starting_with_a = session.query(User).filter(User.name.like('A%')).all()
print([user.name for user in users_starting_with_a])

# Count users
user_count = session.query(User).count()
print(f"Total users: {user_count}")

---

Snippet 8: SQLite Transactions

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

try:
    # Start a transaction
    cursor.execute('INSERT INTO users (name) VALUES (?)', ('Eve',))
    cursor.execute('INSERT INTO users (name) VALUES (?)', ('Frank',))
    conn.commit()  # Commit transaction
except sqlite3.Error as e:
    conn.rollback()  # Rollback on error
    print(f"Transaction failed: {e}")

# Verify data
cursor.execute('SELECT * FROM users')
print(cursor.fetchall())

conn.close()

---

Snippet 9: Using Relationships in SQLAlchemy

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

# Define a new model with relationships
class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship('User', back_populates='posts')

User.posts = relationship('Post', back_populates='user')

# Create tables
Base.metadata.create_all(engine)

# Add a user and post
user = User(name='Alice')
post = Post(title='First Post', user=user)
session.add(user)
session.add(post)
session.commit()

---

Snippet 10: SQLAlchemy Raw SQL Queries

# Execute raw SQL queries with SQLAlchemy
result = session.execute('SELECT * FROM users WHERE name LIKE :name', {'name': 'A%'})
for row in result:
    print(row)

# Close the session
session.close()

---