pip install sqlalchemy

from sqlalchemy import create_engine
# 創(chuàng)建引擎
engine = create_engine('sqlite:///example.db')
# MySQL示例：create_engine('mysql+pymysql://user:password@host:port/dbname')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
# 創(chuàng)建基類
Base = declarative_base()
# 定義用戶模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    def __repr__(self):
        return f"<User(username='{self.username}', email='{self.email}')>"
# 創(chuàng)建數(shù)據(jù)庫(kù)引擎
engine = create_engine('sqlite:///example.db', echo=True)
# 創(chuàng)建表
Base.metadata.create_all(engine)
# 創(chuàng)建會(huì)話
Session = sessionmaker(bind=engine)
session = Session()
# 創(chuàng)建用戶
new_user = User(username='alice', email='alice@example.com')
session.add(new_user)
session.commit()
# 查詢用戶
users = session.query(User).all()
for user in users:
    print(user)
# 關(guān)閉會(huì)話
session.close()

# 單條插入
user = User(username='bob', email='bob@example.com')
session.add(user)
session.commit()
# 批量插入
users = [
    User(username='user1', email='user1@example.com'),
    User(username='user2', email='user2@example.com')
]
session.add_all(users)
session.commit()

# 基本查詢
users = session.query(User).all()
first_user = session.query(User).first()
user_by_id = session.query(User).get(1)
# 條件查詢
active_users = session.query(User).filter(User.is_active == True).all()
user_by_email = session.query(User).filter(User.email == 'alice@example.com').first()
# 復(fù)雜查詢
from sqlalchemy import and_, or_, not_
# AND條件
users = session.query(User).filter(
    and_(User.is_active == True, User.username.like('%alice%'))
).all()
# OR條件
users = session.query(User).filter(
    or_(User.username.like('%alice%'), User.email.like('%example%'))
).all()
# 排序和限制
recent_users = session.query(User).order_by(User.created_at.desc()).limit(10).all()
# 分頁(yè)查詢
page = 1
per_page = 20
users = session.query(User).offset((page-1)*per_page).limit(per_page).all()

# 更新單條記錄
user = session.query(User).filter(User.id == 1).first()
if user:
    user.email = 'newemail@example.com'
    session.commit()
# 批量更新
session.query(User).filter(User.is_active == False).update({
    User.is_active: True
})
session.commit()

# 刪除單條記錄
user = session.query(User).filter(User.id == 1).first()
if user:
    session.delete(user)
    session.commit()
# 批量刪除
session.query(User).filter(User.is_active == False).delete()
session.commit()

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 創(chuàng)建數(shù)據(jù)庫(kù)連接
engine = create_engine('sqlite:///example.db', echo=True)
Base = declarative_base()
# 定義模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)
# 創(chuàng)建表
Base.metadata.create_all(engine)
# 創(chuàng)建會(huì)話工廠
Session = sessionmaker(bind=engine)
# 創(chuàng)建會(huì)話
session = Session()
try:
    # 開(kāi)始事務(wù)
    user = User(username='transaction_user', email='trans@example.com')
    session.add(user)
    post = Post(title='Transaction Post', content='Content here')
    post.author = user
    session.add(post)
    # 提交事務(wù)
    session.commit()
    print("Transaction completed successfully")
except Exception as e:
    # 回滾事務(wù)
    session.rollback()
    print(f"Transaction failed: {e}")
finally:
    session.close()

from contextlib import contextmanager
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 創(chuàng)建數(shù)據(jù)庫(kù)連接
engine = create_engine('sqlite:///example.db', echo=True)
Base = declarative_base()
# 定義模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)
# 創(chuàng)建表
Base.metadata.create_all(engine)
# 創(chuàng)建會(huì)話工廠
Session = sessionmaker(bind=engine)
# 創(chuàng)建會(huì)話
session = Session()
@contextmanager
def get_db_session():
    session = Session()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()
# 使用示例
with get_db_session() as session:
    user = User(username='context_user', email='context@example.com')
    session.add(user)

from sqlalchemy import create_engine
engine = create_engine(
    'mysql+pymysql://user:password@localhost/dbname',
    pool_size=10,           # 連接池大小
    max_overflow=20,        # 最大溢出連接數(shù)
    pool_timeout=30,        # 連接超時(shí)時(shí)間
    pool_recycle=3600       # 連接回收時(shí)間
)