/Users/song/Code/sqlalchemy_mysql_learn00/python_sqlalchemy_mysql/query_table_data.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

'''
@File :  query_table_data.py
@Desc :  查询
'''

# The Python Standard Modules(Library) and Third Modules(Library)
from sqlalchemy.sql import text
from sqlalchemy import and_, or_
# User-defined Modules
from model import Emp, Dept


class QueryTableData:
    """
    查询
    """
    def __init__(self, session):
        self.session = session

    def query_data_01(self):
        # 1 全字段全数据
        rows = self.session.query(Emp).all()
        for row in rows:
            print(row.emp_id, row.emp_name)
        # 2 部分字段第一行数据
        r2 = self.session.query(Emp.emp_id, Emp.emp_name).first()
        print(r2)
        # 3 条件查询
        r3 = self.session.query(Emp.emp_id, Emp.emp_name).filter(Emp.emp_id==1).first()
        print(r3)
        # 4 模糊查询
        r4 = self.session.query(Emp.emp_id, Emp.emp_name).filter(Emp.emp_name.like("%三%")).first()
        print(r4)
        # 5 改变字段名
        r5 = self.session.query(
            Emp.emp_id.label('id'), Emp.emp_name.label('name')
        ).filter(Emp.emp_name.like("%三%")).first()
        print(r5.id, r5.name)
        # 6 使用 sql 语句
        r6 = self.session.query(Emp).from_statement(
            text("SELECT * FROM emp where emp_name=:emp_name")
        ).params(emp_name='张三').first()
        print(r6.emp_id, r6.emp_name)
        # 7 组装
        r7 = self.session.query(Emp).filter(
            text("emp_id<=:emp_id and emp_name=:emp_name")
        ).params(emp_id=1, emp_name='张三').order_by(Emp.emp_id).first()
        print(r7.emp_id, r7.emp_name)
        # 8 in
        rows8 = self.session.query(Emp).filter(Emp.emp_id.in_([1,3,4])).all()
        for row81 in rows8:
            print(row81.emp_id, row81.emp_name)
        # 9 between
        rows9 = self.session.query(Emp).filter(Emp.emp_id.between(1, 3)).all()
        for row91 in rows9:
            print(row91.emp_id, row91.emp_name)
        # 10 and
        r10 = self.session.query(Emp.emp_id, Emp.emp_name).filter(
            and_(Emp.emp_id==1, Emp.emp_name=='张三')
        ).first()
        print(r10)
        # 11 or
        r11 = self.session.query(Emp.emp_id, Emp.emp_name).filter(
            or_(Emp.emp_id==3, Emp.emp_name=='张三')
        ).first()
        print(r11)
        # 12 非
        r12 = self.session.query(Emp).filter(~Emp.emp_id.in_([1,2,4])).first()
        print(r12.emp_id, r12.emp_name)
        # 13 sql
        cursor = self.session.execute(
            text('select * from emp where emp_id=:emp_id'), params={"emp_id": 1}
        )
        r13s = cursor.fetchall()
        for r13 in r13s:
            print(r13)

    def query_data_02(self):
        # 1 多表查询
        # 1.1 方式1：
        r11 = self.session.query(Emp).filter(
            Emp.emp_id.in_(self.session.query(Emp.emp_id).filter_by(emp_name='张三'))
        ).first()
        print(r11.emp_id, r11.emp_name)
        # 1.2 方式2：
        r12 = self.session.query(Emp, Dept).filter(Emp.dept_id == Dept.dept_id).first()
        print(r12[0].emp_name, r12[1].dept_name)
        # 1.3 方式3：
        r13s = self.session.query(Emp).join(Dept).all()
        for r13 in r13s:
            print(r13)
        # 1.4 方式4：
        r14s = self.session.query(Emp).join(Dept, isouter=True).all()
        for r14 in r14s:
            print(r14.emp_name)
        # 2 union
        q1 = self.session.query(Emp.emp_id, Emp.emp_name).filter(Emp.emp_id >= 1)
        q2 = self.session.query(Dept.dept_id, Dept.dept_name).filter(Dept.dept_id >= 1)
        r2s = q1.union(q2).all()
        for r2 in r2s:
            print(r2)
        # 3 union_all
        q1 = self.session.query(Emp.emp_id, Emp.emp_name).filter(Emp.emp_id >= 1)
        q2 = self.session.query(Dept.dept_id, Dept.dept_name).filter(Dept.dept_id >= 1)
        r2s = q1.union_all(q2).all()
        for r2 in r2s:
            print(r2)



    def main(self):
        """
        主函数
        """
        self.query_data_01()
        self.query_data_02()
    

/Users/song/Code/sqlalchemy_mysql_learn00/python_sqlalchemy_mysql/add_table_data.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

'''
@File :  add_table_data.py
@Desc :  新增
'''

# The Python Standard Modules(Library) and Third Modules(Library)

# User-defined Modules
from model import Emp, Dept, Student, Course, Student2Course


class AddTableData:
    """
    新增
    """
    def __init__(self, session):
        self.session = session
    
    def single_table_add_data(self):
        """
        单表新增
        """
        self.session.add(
            Dept(dept_name="开发")
        )
        # self.session.commit()
        self.session.add_all([
            Dept(dept_name="销售"),
            Dept(dept_name="运营"),
            Dept(dept_name="宣传")
        ])
        self.session.commit()
    
    def one_to_many_table_add_data(self):
        """
        一对多表新增
        """
        dept_1 = Dept(dept_name="Python")
        dept_2 = Dept(dept_name='Java')
        dept_1.emp_list = [
            Emp(emp_name='张三'),
            Emp(emp_name='李四')
        ]
        dept_2.emp_list = [
            Emp(emp_name='王二')
        ]
        self.session.add_all([dept_1, dept_2])
        self.session.commit()
    
    def many_to_many_table_add_data(self):
        """
        多对多表新增
        """
        course_1 = Course(course_name='语文')
        course_2 = Course(course_name='数学')
        course_1.student_list = [
            Student(student_name="张三"),
            Student(student_name="李四"),
        ]
        course_2.student_list = [
            Student(student_name="王二")
        ]
        self.session.add_all([course_1, course_2])
        self.session.commit()

    def main(self):
        """
        主函数
        """
        # 单表新增
        self.single_table_add_data()
        # 一对多表新增
        self.one_to_many_table_add_data()
        # 多对多表新增
        self.many_to_many_table_add_data()
    

/Users/song/Code/sqlalchemy_mysql_learn00/python_sqlalchemy_mysql/delete_table_data.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

'''
@File :  delete_table_data.py
@Desc :  删除
'''

# The Python Standard Modules(Library) and Third Modules(Library)

# User-defined Modules
from model import Emp


class DeleteTableData:
    """
    删除
    """
    def __init__(self, session):
        self.session = session
    
    def delete_data(self):
        """
        删除
        """
        self.session.query(Emp).filter(Emp.emp_id==2).delete()
        self.session.commit()
    
    def main(self):
        """
        主函数
        """
        self.delete_data()
    

/Users/song/Code/sqlalchemy_mysql_learn00/python_sqlalchemy_mysql/model.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

'''
@File :  model.py
@Desc :  orm 模型
'''

# The Python Standard Modules(Library) and Third Modules(Library)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, ForeignKey, UniqueConstraint
from sqlalchemy import Integer, String, Date
from sqlalchemy.orm import relationship
# User-defined Modules


Base = declarative_base()

# ================================== 一对多 ================================== #
class Emp(Base):
    __tablename__ = 'emp'

    emp_id = Column(Integer, primary_key=True, autoincrement=True)
    emp_name = Column(String(20), nullable=False)
    birthday = Column(Date)
    sex = Column(Integer)

    # 外键
    dept_id = Column(Integer, ForeignKey('dept.dept_id'))
    # 关系
    dept = relationship('Dept', backref='emp_list')


class Dept(Base):
    __tablename__ = 'dept'

    dept_id = Column(Integer, primary_key=True, autoincrement=True)
    dept_name = Column(String(50), nullable=False)

    __table_args__ = (
        UniqueConstraint('dept_id', 'dept_name', name='dept_id_name'),
    )

# ================================== 多对多 ================================== #
class Student(Base):
    __tablename__ = 'student'

    student_id = Column(Integer, primary_key=True, autoincrement=True)
    student_name = Column(String(20), nullable=False)

    # 关系
    course_list = relationship('Course', secondary='student2course', backref='student_list')

class Course(Base):
    __tablename__ = 'course'

    course_id = Column(Integer, primary_key=True, autoincrement=True)
    course_name = Column(String(20), nullable=False)

class Student2Course(Base):
    __tablename__ = 'student2course'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    student_id = Column(Integer, ForeignKey('student.student_id'))
    course_id = Column(Integer, ForeignKey('course.course_id'))

/Users/song/Code/sqlalchemy_mysql_learn00/python_sqlalchemy_mysql/init_conn.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

'''
@File :  init_conn.py
@Desc :  初始化数据库连接
'''

# The Python Standard Modules(Library) and Third Modules(Library)
from sqlalchemy import create_engine
# User-defined Modules

class InitConn:
    """
    初始化数据库连接
    """
    def __init__(self):
        conn_base_str = "{driver}://{username}:{password}@{host_port}/{database}{parameter}"
        conn_param = {
            'driver': 'mysql+mysqlconnector',
            'username': 'root',
            'password': '123456',
            'host_port': '127.0.0.1:3306',
            'database': 'study',
            'parameter': '?auth_plugin=mysql_native_password&charset=utf8',
        }
        self.engine = create_engine(
            conn_base_str.format_map(conn_param),
            max_overflow=0,     # 超过连接池大小外最多创建的连接
            pool_size=5,        # 连接池大小
            pool_timeout=30,    # 池中没有线程最多等待的时间，否则报错
            pool_recycle=-1     # 多久之后对线程池中的线程进行一次连接的回收（重置）
        )
    
    def get_engine(self):
        """
        获取数据库连接初始化结果
        """
        return self.engine

/Users/song/Code/sqlalchemy_mysql_learn00/python_sqlalchemy_mysql/main.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

'''
@File :  main.py
@Desc :  主测试文件
'''

# The Python Standard Modules(Library) and Third Modules(Library)
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from add_table_data import AddTableData
from delete_table_data import DeleteTableData

# User-defined Modules
from init_conn import InitConn
from model import Base
from query_table_data import QueryTableData
from update_table_data import UpdateTableData

# from query_table_data import QueryTableData
# from add_table_data import AddTableData
# from update_table_data import UpdateTableData
# from delete_table_data import DeleteTableData


if __name__ == '__main__':
    engine = InitConn().get_engine()
    SessionFactory = sessionmaker(bind=engine)
    # 生成 session：
    # 方式一：
    session = scoped_session(SessionFactory)    # 当使用 scoped_session 时，用 session.remove() 关闭 session
    # 方式二：
    # session = SessionFactory()                  # 当使用 scoped_session 时，用 session.close() 关闭 session


    # # 1 删除表
    # Base.metadata.drop_all(engine)
    # 2 创建表
    # Base.metadata.create_all(engine)

    # # 3 查询
    # QueryTableData(session).main()
    # # 4 新增
    # AddTableData(session).main()
    # # 5 修改
    # UpdateTableData(session).main()
    # # 6 删除
    DeleteTableData(session).main()
    
    # 关闭 session
    session.remove()

/Users/song/Code/sqlalchemy_mysql_learn00/python_sqlalchemy_mysql/update_table_data.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-

'''
@File :  update_table_data.py
@Desc :  修改
'''

# The Python Standard Modules(Library) and Third Modules(Library)
import datetime
# User-defined Modules
from model import Emp


class UpdateTableData:
    """
    修改
    """
    def __init__(self, session):
        self.session = session
    
    def update_data(self):
        """
        修改
        """
        self.session.query(Emp).filter(Emp.emp_id==1).update({'birthday': datetime.datetime.now()})
        self.session.commit()
    
    def main(self):
        """
        主函数
        """
        self.update_data()