SQLAlchemy là một thư viện ORM (Object Relational Mapper) mạnh mẽ cho Python, cung cấp một bộ công cụ toàn diện để làm việc với cơ sở dữ liệu quan hệ như SQL Server, MySQL, PostgreSQL, và nhiều hệ quản trị cơ sở dữ liệu khác. Bài viết này sẽ hướng dẫn bạn cách sử dụng SQLAlchemy để tương tác với Microsoft SQL Server, từ thiết lập kết nối ban đầu đến thực hiện các thao tác CRUD (Create, Read, Update, Delete) phức tạp.
Giới thiệu về SQLAlchemy
SQLAlchemy được thiết kế với hai thành phần chính:
-
Core: Cung cấp một SQL abstraction toolkit, cho phép tạo và thực thi các truy vấn SQL thông qua Python mà không cần viết trực tiếp các câu lệnh SQL.
-
ORM (Object Relational Mapper): Cho phép ánh xạ các bảng cơ sở dữ liệu thành các lớp Python và thao tác với dữ liệu như làm việc với các đối tượng Python thông thường.
SQLAlchemy mang lại nhiều lợi ích khi làm việc với SQL Server:
- Tính di động: Mã nguồn có thể dễ dàng chuyển đổi giữa các cơ sở dữ liệu khác nhau
- Bảo mật: Tự động xử lý các vấn đề bảo mật như SQL injection
- Hiệu suất: Tối ưu hóa truy vấn và connection pooling
- Mức độ trừu tượng: Làm việc với dữ liệu ở mức đối tượng thay vì viết SQL thuần
- Hỗ trợ transaction: Quản lý transaction đơn giản và hiệu quả
Cài đặt các thành phần cần thiết
Để làm việc với SQL Server qua SQLAlchemy, bạn cần cài đặt các gói sau:
pip install sqlalchemy
pip install pyodbc
pip install sqlalchemy-pyodbc-mssql
Nếu bạn muốn sử dụng thư viện mới hơn và được khuyến nghị:
pip install pymssql
pip install sqlalchemy[mssql]
Thiết lập kết nối đến SQL Server
1. Tạo URL kết nối
SQLAlchemy sử dụng URL để kết nối đến cơ sở dữ liệu. Dưới đây là cách tạo URL kết nối đến SQL Server:
from sqlalchemy import create_engine, URL
import urllib.parse
connection_string = "mssql+pyodbc://username:password@server_name/database_name?driver=ODBC+Driver+17+for+SQL+Server"
connection_url = URL.create(
"mssql+pyodbc",
username="username",
password="password",
host="server_name",
database="database_name",
query={
"driver": "ODBC Driver 17 for SQL Server",
"TrustServerCertificate": "yes",
"encrypt": "yes",
},
)
params = urllib.parse.quote_plus(
"DRIVER={ODBC Driver 17 for SQL Server};"
"SERVER=server_name;"
"DATABASE=database_name;"
"UID=username;"
"PWD=password;"
"TrustServerCertificate=yes;"
)
connection_url = f"mssql+pyodbc:///?odbc_connect={params}"
2. Tạo Engine
Engine là thành phần trung tâm của SQLAlchemy, đại diện cho kết nối với cơ sở dữ liệu:
from sqlalchemy import create_engine
engine = create_engine(connection_url, echo=True)
Tham số echo=True giúp hiển thị các câu lệnh SQL được tạo ra, rất hữu ích khi gỡ lỗi.
3. Kiểm tra kết nối
try:
with engine.connect() as connection:
result = connection.execute("SELECT @@VERSION")
print(f"Kết nối thành công! Phiên bản SQL Server: {result.scalar()}")
except Exception as e:
print(f"Lỗi kết nối: {e}")
Tạo mô hình dữ liệu (ORM)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.orm import relationship
import datetime
Base = declarative_base()
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
2. Định nghĩa các model
class Customer(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
email = Column(String(100), unique=True)
phone = Column(String(20))
created_at = Column(DateTime, default=datetime.datetime.utcnow)
orders = relationship("Order", back_populates="customer")
def __repr__(self):
return f"<Customer(id={self.id}, name='{self.name}', email='{self.email}')>"
class Order(Base):
__tablename__ = 'orders'
id = Column(Integer, primary_key=True)
customer_id = Column(Integer, ForeignKey('customers.id'))
order_date = Column(DateTime, default=datetime.datetime.utcnow)
total_amount = Column(Float, nullable=False)
status = Column(String(20), default='pending')
customer = relationship("Customer", back_populates="orders")
items = relationship("OrderItem", back_populates="order")
def __repr__(self):
return f"<Order(id={self.id}, customer_id={self.customer_id}, total_amount={self.total_amount})>"
class OrderItem(Base):
__tablename__ = 'order_items'
id = Column(Integer, primary_key=True)
order_id = Column(Integer, ForeignKey('orders.id'))
product_name = Column(String(100), nullable=False)
quantity = Column(Integer, nullable=False)
unit_price = Column(Float, nullable=False)
order = relationship("Order", back_populates="items")
def __repr__(self):
return f"<OrderItem(id={self.id}, order_id={self.order_id}, product_name='{self.product_name}')>"
3. Tạo bảng trong cơ sở dữ liệu
Base.metadata.create_all(engine)
Thao tác CRUD với SQLAlchemy ORM
1. Thiết lập Session
Session là cách SQLAlchemy quản lý các thao tác với cơ sở dữ liệu:
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()
Trong SQLAlchemy 2.0+, bạn có thể sử dụng:
from sqlalchemy.orm import Session
with Session(engine) as session:
pass
2. Thêm dữ liệu (Create)
new_customer = Customer(
name="Nguyễn Văn A",
email="nguyenvana@example.com",
phone="0123456789"
)
session.add(new_customer)
session.add_all([
Customer(name="Trần Thị B", email="tranthib@example.com", phone="0987654321"),
Customer(name="Lê Văn C", email="levanc@example.com", phone="0369874125")
])
session.commit()
3. Truy vấn dữ liệu (Read)
all_customers = session.query(Customer).all()
for customer in all_customers:
print(customer)
customer = session.query(Customer).filter(Customer.email == "nguyenvana@example.com").first()
print(f"Tìm thấy khách hàng: {customer.name}")
from sqlalchemy import or_, and_
customers = session.query(Customer).filter(
or_(
Customer.name.like("Nguyễn%"),
and_(
Customer.email.like("%@example.com"),
Customer.phone.startswith("01")
)
)
).all()
orders_with_customers = session.query(Order, Customer).join(Customer).all()
for order, customer in orders_with_customers:
print(f"Đơn hàng {order.id} thuộc về khách hàng {customer.name}")
from sqlalchemy import func
total_orders = session.query(func.count(Order.id)).scalar()
print(f"Tổng số đơn hàng: {total_orders}")
revenue_by_customer = session.query(
Customer.name,
func.sum(Order.total_amount).label('total_revenue')
).join(Order).group_by(Customer.name).order_by(func.sum(Order.total_amount).desc()).all()
for name, revenue in revenue_by_customer:
print(f"Khách hàng: {name}, Tổng doanh thu: {revenue}")
4. Cập nhật dữ liệu (Update)
customer = session.query(Customer).filter(Customer.email == "nguyenvana@example.com").first()
if customer:
customer.phone = "0123123123"
session.commit()
print(f"Đã cập nhật số điện thoại của khách hàng {customer.name}")
affected_rows = session.query(Order).filter(Order.status == "pending").update(
{"status": "processing"},
synchronize_session=False
)
session.commit()
print(f"Đã cập nhật {affected_rows} đơn hàng từ 'pending' sang 'processing'")
5. Xóa dữ liệu (Delete)
order = session.query(Order).filter(Order.id == 1).first()
if order:
session.delete(order)
session.commit()
print("Đã xóa đơn hàng")
deleted_count = session.query(OrderItem).filter(OrderItem.unit_price < 10000).delete(
synchronize_session=False
)
session.commit()
print(f"Đã xóa {deleted_count} sản phẩm có giá dưới 10.000")
Xử lý transaction và lỗi
1. Sử dụng transaction
from sqlalchemy.orm import Session
try:
with Session(engine) as session:
new_customer = Customer(name="Khách hàng mới", email="khachhang@example.com")
session.add(new_customer)
new_order = Order(customer=new_customer, total_amount=150000, status="new")
session.add(new_order)
session.add_all([
OrderItem(order=new_order, product_name="Sản phẩm A", quantity=2, unit_price=50000),
OrderItem(order=new_order, product_name="Sản phẩm B", quantity=1, unit_price=50000)
])
session.commit()
print("Đã thêm khách hàng và đơn hàng thành công")
except Exception as e:
print(f"Lỗi: {e}")
2. Xử lý commit và rollback thủ công
session = Session()
try:
new_customer = Customer(name="Khách hàng thủ công", email="manual@example.com")
session.add(new_customer)
new_order = Order(customer=new_customer, total_amount=200000)
session.add(new_order)
session.commit()
print("Transaction thành công")
except Exception as e:
session.rollback()
print(f"Transaction thất bại: {e}")
finally:
session.close()
Các tính năng nâng cao của SQLAlchemy
1. Sử dụng SQLAlchemy Core (Expression Language)
Ngoài ORM, bạn có thể sử dụng SQLAlchemy Core để làm việc với cú pháp biểu thức gần hơn với SQL:
from sqlalchemy import Table, MetaData, select, join
metadata = MetaData()
customers = Table('customers', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(100)),
Column('email', String(100)),
Column('phone', String(20))
)
orders = Table('orders', metadata,
Column('id', Integer, primary_key=True),
Column('customer_id', Integer, ForeignKey('customers.id')),
Column('total_amount', Float),
Column('status', String(20))
)
query = select(customers.c.name, orders.c.total_amount).select_from(
join(customers, orders, customers.c.id == orders.c.customer_id)
).where(
orders.c.status == 'completed'
)
with engine.connect() as conn:
result = conn.execute(query)
for row in result:
print(f"Khách hàng: {row.name}, Giá trị đơn hàng: {row.total_amount}")
2. Tạo index và constraints
from sqlalchemy import Index, UniqueConstraint
class Product(Base):
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
sku = Column(String(50), nullable=False)
name = Column(String(100), nullable=False)
price = Column(Float, nullable=False)
category = Column(String(50))
__table_args__ = (
UniqueConstraint('sku', name='uq_product_sku'),
Index('idx_product_name', 'name'),
Index('idx_product_category_price', 'category', 'price')
)
3. Sử dụng lazy loading và eager loading
customer = session.query(Customer).filter(Customer.id == 1).first()
for order in customer.orders:
print(order)
from sqlalchemy.orm import joinedload
customer = session.query(Customer).options(
joinedload(Customer.orders)
).filter(Customer.id == 1).first()
for order in customer.orders:
print(order)
customer = session.query(Customer).options(
joinedload(Customer.orders).joinedload(Order.items)
).filter(Customer.id == 1).first()
for order in customer.orders:
for item in order.items:
print(item)
4. Sử dụng events
from sqlalchemy import event
@event.listens_for(Customer, 'before_insert')
def before_customer_insert(mapper, connection, customer):
print(f"Chuẩn bị thêm khách hàng: {customer.name}")
customer.email = customer.email.lower()
@event.listens_for(Customer, 'after_insert')
def after_customer_insert(mapper, connection, customer):
print(f"Đã thêm khách hàng: {customer.name} với ID = {customer.id}")
Các thực hành tốt nhất và mẹo khi sử dụng SQLAlchemy với SQL Server
1. Sử dụng connection pooling
engine = create_engine(
connection_url,
pool_size=10,
max_overflow=20,
pool_timeout=30,
pool_recycle=1800
)
2. Sử dụng bulk operations cho hiệu suất cao
products = [
Product(sku=f"PRD-{i}", name=f"Sản phẩm {i}", price=10000 + i * 1000, category="Electronics")
for i in range(1, 1001)
]
session.bulk_save_objects(products)
session.commit()
3. Quản lý migration với Alembic
Alembic là công cụ migration được phát triển bởi tác giả của SQLAlchemy:
pip install alembic
alembic init migrations
Trong file env.py của Alembic, cấu hình metadata:
from sqlalchemy import engine_from_config, pool
from models import Base
target_metadata = Base.metadata
Tạo migration và áp dụng:
alembic revision --autogenerate -m "Create initial tables"
alembic upgrade head
4. Sử dụng stored procedures
from sqlalchemy import text
with engine.connect() as conn:
result = conn.execute(
text("EXEC GetCustomerOrders :customer_id"),
{"customer_id": 1}
)
for row in result:
print(row)
Kết luận
SQLAlchemy cung cấp một cách mạnh mẽ và linh hoạt để làm việc với SQL Server từ Python. Bằng cách sử dụng ORM, bạn có thể tập trung vào logic nghiệp vụ của ứng dụng thay vì viết các câu lệnh SQL thủ công. Tuy nhiên, SQLAlchemy cũng rất linh hoạt, cho phép bạn viết truy vấn SQL thuần khi cần thiết.
Ngoài ra, SQLAlchemy còn có nhiều tính năng nâng cao như relationship, eager loading, connection pooling, và event listeners, giúp bạn xây dựng các ứng dụng có hiệu suất cao và dễ bảo trì.
Khi làm việc với SQL Server, hãy nhớ cấu hình các tham số kết nối phù hợp và sử dụng các trình điều khiển như pyodbc hoặc pymssql cho hiệu suất tốt nhất.
Bạn đã có kinh nghiệm sử dụng SQLAlchemy chưa? Thư viện ORM này đã giúp ích như thế nào trong các dự án của bạn? Hãy chia sẻ trong phần bình luận nhé!
