diff --git a/sqlalchemy_1.4/db_schema.py b/sqlalchemy_1.4/db_schema.py index fa0039e..a3eafb5 100644 --- a/sqlalchemy_1.4/db_schema.py +++ b/sqlalchemy_1.4/db_schema.py @@ -1,6 +1,7 @@ import sqlalchemy -from sqlalchemy.orm import declarative_base, sessionmaker, Session # declarative 声明式 -from sqlalchemy import create_engine, Column, Integer, String +from sqlalchemy.orm import declarative_base, sessionmaker, Session, relationship, backref # declarative 声明式 +from sqlalchemy import create_engine, Column, Integer, String, DATE, Enum, desc, or_, and_, update, case, union, \ + union_all # 创建数据引擎 HOST = "home.rogersun.cn" @@ -9,13 +10,14 @@ PWD = "Sxzgx1209" PORT = "3306" DB = "orm_sqlalchemy" -URI = f"mysql+pymysql://{USER}:{PWD}@{HOST}:{PORT}/{DB}" +URI = f"mysql+pymysql://{USER}:{PWD}@{HOST}:{PORT}/{DB}?charset=utf8" engine = create_engine(URI, echo=True) # 创建session -session = sessionmaker(engine) +SessionClass = sessionmaker(bind=engine) +session = SessionClass() -# with session +# # with Session # with Session(engine) as db_session: # session.begin() # try: @@ -31,5 +33,130 @@ session = sessionmaker(engine) Base = declarative_base() +# 创建表结构 class User(Base): __tablename__ = 'user' + id = Column(Integer, primary_key=True) + name = Column(String(32)) + password = Column(String(64)) + + def __repr__(self): + return "<%s name:%s>" % (self.id, self.name) + + +class Student(Base): + __tablename__ = 'student' + id = Column(Integer, primary_key=True) + name = Column(String(32), nullable=False) + gender = Column(Enum('m', 'f'), nullable=False) + date = Column(DATE, nullable=False) + + def __repr__(self): + return f"<{self.id}-{self.name}-{self.gender}-{self.date}>" + + +# 生成实体表 +# Base.metadata.create_all(engine) + +# insert +# 创建数据 +user_data_1 = User(name="roger", password="123321") +user_data_2 = User(name="honey", password='123456') +# +# # 插入数据 +# session.add(user_data_1) +# session.add(user_data_2) +# +# # 提交数据 +# session.commit() +# insert完整示例 +# try: +# session.add(user_data_1) +# except Exception as e: +# print(e) +# session.rollback() +# else: +# session.commit() + + +# select +# result = session.query(User).all() # all +# result = session.query(User).filter_by(name='roger').all() # filter_by +# result = session.query(User).filter(User.id == 2).all() # filter +# result = session.query(User).filter(User.id >= 2).first() # first +# result = session.query(User).filter(User.id >= 2).order_by(User.id.desc()).all() # order_by desc +# result = session.query(User).filter(User.id >= 2).filter(User.name.like('%one%')).all() # like +# result = session.query(User).filter(User.id.in_([1, 2, 3])).all() # in +# # in startswith +# result = session.query(User).filter( +# User.id.in_(session.query(User.id).filter(User.name.startswith('he'))) +# ).all() +# result = session.query(User).filter(User.id.notin_([1, 2])).all() # notin +# result = session.query(User).filter(User.password is None).all() # is None +# result = session.query(User).filter(User.password.is_(None)).all() # is_(None) +# result = session.query(User).filter(User.password is not None).all() # is not None +# result = session.query(User).filter(User.password.isnot(None)).all() # isnot(None) +# result = session.query(User).filter(or_(User.id > 2, User.name.ilike('Rog%'))).all() # or_ +# result = session.query(User).filter(and_(User.id >= 2, User.name.endswith('ei'))).all() # and_ endswith +# result = session.query(User.name.label('user_name'), User.password).filter(User.id >= 2).all() # label print(result[0].user_name) +# result = session.query(User).filter(User.id > 1).count() # count +# try: +# result = session.query(User).filter(User.id > 1).one() # one 当仅查询到一个结果时返回,否则抛出异常 +# except Exception as e: +# print(e) +# result = session.query(User).filter(User.id == 1).one_or_none() # one_or_none 当查询结果仅一条时显示返回结果,当没有结果时返回None,否则抛出异常 +# result = session.query(User).filter(User.id == 1).scalar() # 调用one方法,并在成功时返回行的第一列 +result = session.query(User).filter(User.id > 1).group_by(User.password) +# print(result[0].name, result[0].password) +print(result) # 定义 __repr__ 后可以打印 + +# join +# filter形式 +# result = session.query(User, Student).filter(User.name == Student.name).all() +# print(result[0].User.name, result[0].User.password, result[0].Student.name, result[0].Student.gender, +# result[0].Student.date) +# +# # join形式 无外键关联的情况 +# result = session.query(User).join(Student, User.name == Student.name).all() +# print(result[0].name, result[0].password) + +# join有外键关联的情况(Todo) + + +# union +# union +# q1 = session.query(User.name).filter(User.id > 1) +# q2 = session.query(Student.name).filter(Student.id <= 2) +# result = q1.union(q2).all() +# print(result) + +# union_all +# q1 = session.query(User.id, User.name) +# q2 = session.query(Student.id.label('id'), Student.name.label('name')) +# result = q1.union_all(q2).filter(User.id <= 2) +# print(dict(result)) +# for r in result: +# print(r) + + +# # update +# 一般形式 +# data = session.query(User).filter(User.id == 1).first() +# print(data) +# data.name = 'rogersun' +# session.commit() +# 连写形势 +# data = session.query(User).filter(User.id == 2).update({'name': 'honeyhoney'}) +# session.commit() +# 批量更新 +# data = session.query(User).filter(User.password.isnot(None)).update({User.password: '123321'}) +# print(data) +# session.commit() + +# delete +result = session.query(User).filter(User.name == 'delete').delete() +print(result) +session.commit() + +# 关闭session +session.close()