python sqlalchemy 進行 mysql 數據庫操作

1. 進行mysql數據庫的創建，如果已經存在，就相當於進行數據庫的連接操作

from sqlalchemy import create_engine 
from sqlalchemy.ext.declarative import declarative_base 
from sqlalchemy import Column, Integer, String, Date 
from sqlalchemy.orm import sessionmaker 


# 創建連接
engine = create_engine('mysql+pymysql://root:123456@localhost/oldboydb',
                       encoding='utf-8')

Base = declarative_base()  # 生成orm基類

class User(Base): 
      __tablename__ = 'user'   # table的名字
    id = Column(Integer, primary_key=True)  # 創建id屬性
    name = Column(String(32))  # 創建name屬性
    password = Column(String(64)) # 創建密碼屬性

    def __repr__(self):  # 用於進行查找時的數據返回
        return '<%s name :%s>' %(self.id, self.name)

Base.metadata.create_all(engine) # 進行指令的調用，即生成table 

2. 進行數據的查詢，包括條件查詢和多條件查詢

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Date
from sqlalchemy.orm import sessionmaker


# 創建連接
engine = create_engine('mysql+pymysql://root:123456@localhost/oldboydb',
                       encoding='utf-8')

Base = declarative_base()  # 生成orm基類

class User(Base):  # User繼承了上述的操作

    __tablename__ = 'student_user'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    password = Column(String(64))

    def __repr__(self): # 返回查詢到的信息
        return '<id: %s name: %s password: %s>' %(self.id, self.name, self.password)

# 執行上述的操作
Base.metadata.create_all(engine)
Session_class = sessionmaker(bind=engine)  # 進行數據庫的連接
Session = Session_class() # 生成session 實例

# 進行數據查詢 .all()輸出所有的檢索結果，.first() 輸出第一條的結果
data = Session.query(User).filter_by(name='alex').first()
print(data)
data = Session.query(User).filter(User.name == 'rain').first()
print(data)

# 多條件查詢
data = Session.query(User).filter(User.name=='rain').filter(User.id > 1).first()
print(data)

3. 進行用戶屬性的修改

# 進行數據的條件修改
data = Session.query(User).filter(User.name=='rain').first() # 查找出數據的類
data.name = 'Tom' # 對類屬性進行修改
Session.commit() # 結果的提交
print(data.id)  # 打印查找結果的id信息

4. 添加新的用戶信息

fake_user = User(name='Jack', password='123456')  # 設置用戶的名字和密碼
Session.add(fake_user) # 將數據添加到表內

5. 進行用戶信息的刪除

data = Session.query(User).filter(User.name=='Jack').first()  # 查找用戶信息
Session.delete(data)  # 刪除用戶 
Session.commit() # 提交結果

6. 數據的回滾操作

data = Session.query(User).filter(User.name=='Jack').first() # 查找符合條件的數據
Session.delete(data) # 數據刪除
Session.rollback()  # 數據回滾，取消上述操作
Session.commit() # 提交結果

7. 數據統計

print(Session.query(User).filter(User.name.in_(['alex', 'Jack'])).count())  # 統計名字是'alex'或者'Jack'的次數

# 2 

8. 根據名字進行數據分組

from sqlalchemy import func
data = Session.query(User.name, func.count(User.name)).group_by(User.name).all()
print(data)

# [('alex', 1), ('Tom', 1), ('Jack', 1)]

 

9. 定義數據庫的student表，進行連表查詢

# 定義Student類關聯student table
class Student(Base):

    __tablename__ = 'student'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    register_data = Column(Date)
    sex = Column(String(32))


    def __repr__(self):  # 返回查詢到的信息
        return '<id: %s name: %s register_data: %s>' % (self.id, self.name, self.register_data)


data = Session.query(User, Student).filter(User.name==Student.name).all()

print(data)

 

10. 將兩個表的屬性進行串聯, 這里將student的id屬性，有上課記錄study_record的stu_id 屬性進行關聯，構造一個學生的上課記錄

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Date, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship


# 創建連接
engine = create_engine('mysql+pymysql://root:123456@localhost/oldboydb',
                       encoding='utf-8')

Base = declarative_base()  # 生成orm 基類

# 構造student的類
class Student(Base):

    __tablename__ = 'student'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    register_data = Column(Date, nullable=False)
    sex = Column(String(32))

    def __repr__(self):
        return '<id: %s name: %s sex: %s>'%(self.id, self.name, self.sex)


class StudyRecord(Base):

    __tablename__ = 'study_record'
    id = Column(Integer, primary_key=True)
    day = Column(Integer, nullable=False)
    status = Column(String(32), nullable=False)
    stu_id = Column(Integer, ForeignKey('student.id'))  # 構建外鍵

    student = relationship('Student', backref='my_study_record')

    def __repr__(self):
        return '<%s day:%s status:%s>'%(self.student.name, self.day, self.status)

Base.metadata.create_all(engine)

Session_class = sessionmaker(bind=engine)

session = Session_class()


s1 = Student(name='Alex', register_data='2015-04-21')
s2 = Student(name='Jack', register_data='2016-04-21')
s3 = Student(name='Rain', register_data='2017-04-21')
s4 = Student(name='Eric', register_data='2018-04-21')

study_obj1 = StudyRecord(day=1, status='YES', stu_id=1)
study_obj2 = StudyRecord(day=2, status='NO', stu_id=1)
study_obj3 = StudyRecord(day=3, status='YES', stu_id=1)
study_obj4 = StudyRecord(day=1, status='YES', stu_id=2)

# 進行數據的批量插入
session.add_all([s1, s2, s3, s4, study_obj1, study_obj2, study_obj3, study_obj4])

stu_obj = session.query(Student).filter(Student.name=='alex').first()
print(stu_obj.my_study_record) # 找出student='alex' 對應的學習記錄

 

11. 一張表里的兩個屬性對應一張表的屬性，就好比商品的收獲地址和發貨地址都對應的是地址數據庫

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Date, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship



# 創建連接
engine = create_engine('mysql+pymysql://root:123456@localhost/oldboydb',
                       encoding='utf-8')

Base = declarative_base()


class Customer(Base):

    __tablename__ = 'customer'
    id = Column(Integer, primary_key=True)
    name = Column(String(64))

    billing_address_id = Column(Integer, ForeignKey('address.id'))
    shipping_address_id = Column(Integer, ForeignKey('address.id'))

    billing_address = relationship('Address', foreign_keys = [billing_address_id])
    shipping_address = relationship('Address', foreign_keys = [shipping_address_id])


class Address(Base):

    __tablename__ = 'address'
    id = Column(Integer, primary_key=True)
    street = Column(String(64))
    city = Column(String(64))
    state = Column(String(64))

    def __repr__(self):
        return self.street

Base.metadata.create_all(engine)

Session_class = sessionmaker(bind=engine)

session = Session_class()

stu_obj = session.query(Customer).filter(Customer.name=='computer').first()
print(stu_obj.shipping_address) # 找出student='alex' 對應的學習記錄

 

12. 構建多對多的表格類型，即一本書可以對應多個作者，一個作者可以對應多本書

orm_m2m.py 

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Date, ForeignKey, Table
from sqlalchemy.orm import sessionmaker, relationship

# 創建連接
engine = create_engine('mysql+pymysql://root:lishentao22@localhost/oldboydb?charset=utf8',
                       )

Base = declarative_base()


book_m2m_author = Table('book_m2m_author', Base.metadata,
                        Column('book_id', ForeignKey('books.id')),
                        Column('author_id', ForeignKey('authors.id')))


class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    name = Column(String(64))
    pub_date = Column(Date)
    authors = relationship('Author', secondary=book_m2m_author, backref='books')


    def __repr__(self):
        return self.name


class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))

    def __repr__(self):
        return self.name

Base.metadata.create_all(engine)

orm_m2m_api.py

from day12 import orm_m2m
from sqlalchemy.orm import sessionmaker



Session_class = sessionmaker(bind=orm_m2m.engine)
session = Session_class()

# b1 = orm_m2m.Book(name='learn python with Alex', pub_date='2014-06-02')
# b2 = orm_m2m.Book(name='learn Zhangbility with Alex', pub_date='2015-02-06')
# b3 = orm_m2m.Book(name='go alex to taiguo', pub_date='2014-02-05')
#
# a1 = orm_m2m.Author(name='Alex')
# a2 = orm_m2m.Author(name='Jack')
# a3 = orm_m2m.Author(name='Rain')
#
# b1.authors = [a1, a3]
# b3.authors = [a1, a2, a3]
#
# session.add_all([b1, b2, b3, a1, a2, a3])
# session.commit()


author_obj = session.query(orm_m2m.Author).filter(orm_m2m.Author.name=='alex').first()
# session.delete(author_obj)  # 刪除這個作者

book_obj = session.query(orm_m2m.Book).filter(orm_m2m.Book.id==14).first()
book_obj.authors.remove(author_obj)

session.commit()