當前位置：首頁 >

根据数据库表gengxin实体类_Python学习第四十八天记录打call：SQLALchemy操作MySQL关系型数据库...

發布時間：2025/3/19 38 豆豆

生活随笔收集整理的這篇文章主要介紹了根据数据库表gengxin实体类_Python学习第四十八天记录打call：SQLALchemy操作MySQL关系型数据库... 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

1.SQLALchemy使用

安裝

pip install sqlalchemy；

SQLAlchemy是Python編程語言下的一款開源軟件，是PythonSQL工具包和對象關系映射器，它為應用程序開發人員提供了SQL的全部功能和靈活性

2.什么是ORM

ORM（Object Relational Mapper）就是把數據庫表的行與相應的對象建立關聯，互相轉換；
目前，最知名的Python ORM是SQLAlchemy和SQLobject；

3.數據庫操作

創建表使用create_all()方法，刪除表使用drop_all()方法，我們一起來看一下如何創建和刪除表

# 導入引擎模塊 from sqlalchemy import create_engine # 導入基類模塊 from sqlalchemy.ext.declarative import declarative_base # 導入字段類 from sqlalchemy import Column, Integer, String # 導入會話模塊 from sqlalchemy.orm import sessionmaker# 實體類的基類 Base = declarative_base()# 實體類 class Teacher(Base):'''實體類的創建有兩個方面的用處：1：如果數據庫中沒有表，那么可以使用實體類創建2：如果數據庫中有表，實體類可以映射表的結構，對表的CRUD操作'''# 表名必須指定__tablename__ = 'teacher'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(255))age = Column(Integer)# 打印實例的使用可以發現，數據也是保存在實體類實例的__dict__中def __repr__(self):# print(self.__dict__)return "id='%s', name='%s', age='%s" % (self.id, self.name, self.age)__str__ = __repr__# 創建連接引擎 host = 'localhost' port = 3306 username = 'zengzeng' password = '123456' db = 'XKD_Python_Course' connect_str = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(username, password, host, port,db) engine = create_engine(connect_str, echo=True)# 創建表 Base.metadata.create_all(engine)# 刪除表 # Base.metadata.drop_all(engine)

我們現在命令工具中查看一下數據庫中有沒有我們想要的teacher表，上一篇文章中講過，先登錄mysql：mysql -uzengzeng -p123456，然后進入我們要使用的數據庫：use XKD_Python_Course，我們先通過：show tables;查看一下數據庫中的表，發現沒有teacher表
那現在就可以執行代碼，創建數據庫了，創建好后我們可以在命令行查看一下：show tables;，發現teacher表已經存在了，創建表成功yes!

查看表

想要刪除剛剛創建的表，可以使用drop_all()方法，執行代碼，再查看數據庫表，就可以看teacher表已經被刪除了

Base.metadata.drop_all(engine)

插入數據使用session.add_all()方法；

from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String from sqlalchemy.orm import sessionmakerBase = declarative_base() class Teacher(Base):__tablename__ = 'teacher'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(255))age = Column(Integer)def __repr__(self):# print(self.__dict__)return "id='%s', name='%s', age='%s" % (self.id, self.name, self.age)__str__ = __repr__host = 'localhost' port = 3306 username = 'zengzeng' password = '123456' db = 'XKD_Python_Course' connect_str = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(username, password, host, port,db) engine = create_engine(connect_str, echo=True)# 創建會話，用于提交數據 Session = sessionmaker(bind=engine) session = Session()# 創建多行行實例，給表添加數據 try:lst = []for i in range(10):teacher = Teacher()teacher.name = 'zengzeng' + str(i)teacher.age = 20 + ilst.append(teacher)print(teacher)# session.add(student) 可以添加一行記錄，也可以添加多行記錄# 注意：這里將行記錄實例添加到session，不會提交，需要手動提交session.add_all(lst)except Exception as e:print('~~~~~~~~~~~'*200)session.rollback()print(e) finally:session.commit()

我們執行代碼，然后去命令工具查看一下表是否插入數據：select * from teacher;

插入數據

查詢操作使用session.query()方法，迭代查詢；

from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String from sqlalchemy.orm import sessionmakerBase = declarative_base() class Teacher(Base):__tablename__ = 'teacher'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(255))age = Column(Integer)def __repr__(self):# print(self.__dict__)return "id='%s', name='%s', age='%s" % (self.id, self.name, self.age)__str__ = __repr__host = 'localhost' port = 3306 username = 'zengzeng' password = '123456' db = 'XKD_Python_Course' conn_str = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(username, password, host, port,db) engine = create_engine(conn_str, echo=False)Session = sessionmaker(bind=engine) session = Session()teacher_obj = session.query(Teacher) for teacher in teacher_obj:print(teacher) # 返回結果：{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x1063125f8>, 'age': 20, 'name': 'nihao0', 'id': 1} print('*'*300)# 直接返回實例對象 teacher = session.query(Teacher).get(4) print(teacher) # 返回結果：{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x1075fd400>, 'age': 23, 'name': 'nihao3', 'id': 4} print(teacher.id) # 返回 aobama print(teacher.name) print(teacher.age) print('*'*300)# 返回的是可迭代對象 teacher_results = session.query(Teacher).filter(Teacher.id == 5) for teacher in teacher_results: # 拿到student實例對象print(teacher)

修改操作：session.update()；

where條件查詢

from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, DATE, Enum, ForeignKey import enum from sqlalchemy.orm import sessionmaker Base = declarative_base() class My_Enum(enum.Enum):M = 'M'F = 'F'# 實體類 class Emploee(Base):'''+------------+---------------+------+-----+---------+-------+| Field | Type | Null | Key | Default | Extra |+------------+---------------+------+-----+---------+-------+| emp_no | int(11) | NO | PRI | NULL | || birth_date | date | NO | | NULL | || first_name | varchar(14) | NO | | NULL | || last_name | varchar(16) | NO | | NULL | || gender | enum('M','F') | NO | | NULL | || hire_date | date | NO | | NULL | |+------------+---------------+------+-----+---------+-------+'''__tablename__ = 'employees'emp_no = Column(Integer, primary_key=True, nullable=False)birth_date = Column(DATE, nullable=False)first_name = Column(String(14), nullable=False)last_name = Column(String(16),nullable=False)gender = Column(Enum(My_Enum), nullable=False)hire_date = Column(DATE, nullable=False)def __repr__(self):return "emp_no='%s', birth_date='%s', first_name='%s', last_name='%s', gender='%s', hire_date='%s'" % (self.emp_no, self.birth_date, self.first_name, self.last_name, self.gender, self.hire_date)__str__ = __repr__# 創建連接引擎 host = 'localhost' port = 3306 username = 'zengzeng' password = '123456' db = 'XKD_Python_Course' conn_str = 'mysql+mysqldb://{}:{}@{}:{}/{}'.format(username, password, host, port,db) engine = create_engine(conn_str, echo=False)# 創建表 Base.metadata.create_all(engine)# 創建會話，用于提交數據 Session = sessionmaker(bind=engine) session = Session()# # 簡單的where條件查詢過濾，返回可迭代對象, AND 取與 emploees = session.query(Emploee).filter(Emploee.emp_no < 10010).filter(Emploee.gender == 'M') emploees = session.query(Emploee).filter((Emploee.emp_no > 10010) & (Emploee.gender == 'F'))# OR 取并 emploees = session.query(Emploee).filter((Emploee.emp_no > 10010) | (Emploee.gender == 'F'))# NOT 取反 emploees = session.query(Emploee).filter(~(Emploee.emp_no > 10010))# in emploees = session.query(Emploee).filter(Emploee.emp_no.in_([10010, 10011, 10012]))# not in emploees = session.query(Emploee).filter(~Emploee.emp_no.in_([10010, 10011, 10012]))# like ，like可以忽略大小寫進行模式匹配 emploees = session.query(Emploee).filter(Emploee.last_name.like('B%'))for emploee in emploees:print(emploee)

order排序

asc()：升序；

例如： emploees = session.query(Emploee).filter(Emploee.last_name.like('B%')).order_by(Emploee.emp_no.asc())

desc()：降序；

例如： emploees=session.query(Emploee).filter(Emploee.last_name.like('B%')).order_by(Emploee.emp_no.desc())

多字段排序：在語句里面進行多次排序；

聚合與分組

list()：轉化為列表；
count()：聚合count(*)查詢；
all()：轉化為列表；
limit().one()：查詢首行；

emploees = session.query(Emploee) print(list(emploees)) # 轉化為列表 print(emploees.count()) # 聚合count(*)查詢 print(emploees.all()) # 轉化為列表 print(emploees.limit(1).one()) # 查詢首行

max()：返回最大值；
min()：返回最小值；
avg()：返回平均值；

emploees = session.query(func.max(Emploee.emp_no)) emploees = session.query(func.min(Emploee.emp_no)) emploees = session.query(func.avg(Emploee.emp_no))