pip install sqlalchemy pymysql

from sqlalchemy import create_engine
# 替換為你的MySQL數(shù)據(jù)庫(kù)信息
username = 'your_mysql_username'
password = 'your_mysql_password'
host = 'your_mysql_host'  # 例如：'localhost' 或 '127.0.0.1'
port = 'your_mysql_port'  # 通常是 3306
database = 'your_database_name'
# 創(chuàng)建連接引擎
engine = create_engine(f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}')

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

Base.metadata.create_all(engine)

from sqlalchemy.orm import sessionmaker
# 創(chuàng)建會(huì)話
Session = sessionmaker(bind=engine)
session = Session()
# 添加新用戶(hù)
new_user = User(name='John Doe', email='john.doe@example.com')
session.add(new_user)
session.commit()
# 關(guān)閉會(huì)話
session.close()

session = Session()
# 查詢(xún)所有用戶(hù)
users = session.query(User).all()
for user in users:
    print(f"User ID: {user.id}, Name: {user.name}, Email: {user.email}")
# 關(guān)閉會(huì)話
session.close()

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String)
    email = Column(String)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 連接到MySQL數(shù)據(jù)庫(kù)（請(qǐng)?zhí)鎿Q為你的數(shù)據(jù)庫(kù)信息）
engine = create_engine('mysql+pymysql://user:password@localhost/dbname')
Session = sessionmaker(bind=engine)
session = Session()

try:
    # 開(kāi)始一個(gè)新的事務(wù)
    session.begin()
    # 創(chuàng)建新用戶(hù)對(duì)象
    user1 = User(name='Alice', email='alice@example.com')
    user2 = User(name='Bob', email='bob@example.com')
    # 添加到會(huì)話中
    session.add(user1)
    session.add(user2)
    # 提交事務(wù)，將所有更改保存到數(shù)據(jù)庫(kù)
    session.commit()
    print("Users added successfully.")
except Exception as e:
    # 如果在添加用戶(hù)過(guò)程中發(fā)生錯(cuò)誤，則回滾事務(wù)
    session.rollback()
    print(f"An error occurred: {e}")
finally:
    # 關(guān)閉會(huì)話
    session.close()

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
Base = declarative_base()
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    orders = relationship("Order", back_populates="user")
class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    product = Column(String)
    quantity = Column(Integer)
    user = relationship("User", back_populates="orders")

from sqlalchemy.orm import joinedload
# 加載所有用戶(hù)的訂單信息
users_with_orders = session.query(User).options(joinedload(User.orders)).all()
for user in users_with_orders:
    print(f"User: {user.name}")
    for order in user.orders:
        print(f"  Order: {order.product}, Quantity: {order.quantity}")

from sqlalchemy import func
# 按用戶(hù)分組，并計(jì)算每個(gè)用戶(hù)的訂單數(shù)量
order_count_by_user = session.query(User.id, User.name, func.count(Order.id).label('order_count')).\
    join(Order).group_by(User.id, User.name).all()
for user_id, user_name, order_count in order_count_by_user:
    print(f"User ID: {user_id}, Name: {user_name}, Order Count: {order_count}")

from sqlalchemy import func, select
# 計(jì)算平均訂單數(shù)量作為子查詢(xún)
avg_order_quantity = select([func.avg(Order.quantity).label('avg_quantity')]).select_from(Order).alias()
# 查詢(xún)訂單數(shù)量超過(guò)平均值的用戶(hù)及其訂單信息
users_above_avg = session.query(User, Order.product, Order.quantity).\
    join(Order).filter(Order.quantity > avg_order_quantity.c.avg_quantity).all()
for user, product, quantity in users_above_avg:
    print(f"User: {user.name}, Product: {product}, Quantity: {quantity}")

# 查詢(xún)名字以“A”開(kāi)頭的用戶(hù)，且訂單中包含“apple”產(chǎn)品的用戶(hù)信息
users_with_apple = session.query(User).join(Order).\
    filter(User.name.startswith('A')).\
    filter(Order.product.contains('apple')).\
    distinct().all()  # 使用distinct()確保結(jié)果中的用戶(hù)不重復(fù)
for user in users_with_apple:
    print(f"User: {user.name}")