1. 程式人生 > >SQLAlchemy基本使用,建立表,增刪改查

SQLAlchemy基本使用,建立表,增刪改查

基礎語法

  • 建立連線
from sqlalchemy import create_engine
# 寫法1
engine = create_engine("postgresql://scott:tiger@localhost/test?charset=utf8")

# 寫法2
engine = create_engine("mysql+pymysql://root:[email protected]/test",encoding='latin1', echo=True")

URL的字串形式是 dialect[+driver]://user:password@host/dbname[?key=value..]

,在這裡 dialect是一個數據庫的名稱,如mysqloraclepostgresql等等,和driver一個DBAPI的名稱,諸如 psycopg2pyodbccx_oracle等。或者,該URL可以是一個例項。

  • 建立回話,

    用來與資料庫交談,ORM對資料庫的“處理”是Session。當我們第一次設定應用程式時,在與create_engine()語句相同的級別上,我們定義一個Session類,它將作為新Session 物件的工廠

from sqlalchemy import sessionmaker
Session = sessionmaker(bind=engine)
db_session = Session()

SQLAlchemy建立表,及簡單的CRUD

1.建立表

# 1. 建立連線
from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

# 2. 建立基類, 相當於 django orm中的model.model
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

from sqlalchemy import Column,Integer,String
# 3. 建立表模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer,primary_key=True,autoincrement=True)
    name = Column(String(32))

    def __repr__(self):
        return self.name

Base.metadata.create_all(engine)

2.單表的CRUD

增加資料

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
db_session = Session()

# 1. 新增單條資料
# u = User(name='使用者1')
# db_session.add(u)

# 2. 新增多條資料
db_session.add_all([
    User(name='使用者2'),
    User(name='使用者3'),
])

# 結束記得提交,資料才能儲存在資料庫中
db_session.commit()
# 關閉會話
db_session.close()

查詢資料

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
db_session = Session()

# 查詢使用者表中所有使用者
user_all = db_session.query(User).all()

# where條件查詢
user = db_session.query(User).filter(User.id>=2).all()
# 取出一條資料
user = db_session.query(User).filter(User.id>=2).first()

db_session.close()

修改資料

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
db_session = Session()


# 更新單條資料
res = db_session.query(User).filter(User.id==20).update({"name":"新使用者"})
db_session.commit()
db_session.close()

# 更新多條資料
res2 = db_session.query(User).filter(User.id<=20).update({"name":"修改使用者"})
print(res)  # 這裡結果的數字是更新的行數

db_session.commit()
db_session.close()

刪除資料

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
db_session = Session()

res = db_session.query(User).filter(User.id==20).delete()

db_session.commit()
db_session.close()

複雜條件查詢

# 1.and_ or_ 條件查詢
from sqlalchemy.sql import and_,or_
ret = db_session.query(User).filter(and_(User.id>3,User.name=='xxx')).all()
ret2 = db_session.query(User).filter(or_(User.id<2,User.name=='xxx')).all()

# 2. 查詢所有資料
res = db_session.query(User).all()

# 3. 查詢資料,指定查詢資料列加入別名
r3 = db_session.query(User.name.label('username'),User.id).firset()
print(r2.id,r2.username)

# 4. 篩選條件格式
r4 = db_session.query(User).filte(User.name='xxx').all()
r5 = db_session.query(User).filter_by(name='xxx').first()

# 6. 原生sql查詢
r6 = db_session.query(User).from_statement(text("select * from User where name=:name")).params(name='xxx').all()

# 7. 篩選查詢列, query的時候不再使用User ORM物件,而是使用User.name來對內容進行選取
user_list = db_session.query(User.name).all()
print(user_list)

for row in user_list:
    print(row.name)

# 8. 複雜查詢
from sqlalchemy.sql import text
user_list = db_session.query(User).filter(text('select * from User id<:value and name=:name')).params(value=3,name='xxx')

# 9. 排序
user_list = db_session.query(User).order_by(User.id.desc()).all()

# 10. between 查詢大於1小於3的
ret = session.query(User).filter(User.id.between(1, 3), User.name == 'xxx').all() 

# 11. in 只查詢id等於1,3,4的
ret = session.query(User).filter(User.id.in_([1,3,4])).all() 

# 12. 取反 查詢不等於1,3,4的
ret = session.query(User).filter(~User.id.in_([1,3,4])).all() 

# 13. 子查詢
ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='xxx'))).all() 

# 14. 萬用字元
ret = db_session.query(User).filter(~User.name.like('e%')).all()

# 15. 限制
ret = db_session.query(User)[1:2]

# 16. 分組
from sqlalchemy.sql import func
ret = db_session.query(User).group_by(User.extra).all()
ret = db_session.query(
    func.max(User.id),
    func.sum(User.id),
    func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all()

複雜修改資料

#高階版更新操作
from my_create_table import User,engine
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)
db_session = Session()

#直接修改
db_session.query(User).filter(User.id > 0).update({"name" : "xxx"})

#在原有值基礎上新增 方式1, 名字後面加字尾
db_session.query(User).filter(User.id > 0).update({User.name: User.name + "字尾"}, synchronize_session=False)

#在原有值基礎上新增 方式2, 年齡加一歲
db_session.query(User).filter(User.id > 0).update({"age": User.age + 1}, synchronize_session="evaluate")
db_session.commit()

3.一對多的操作 外來鍵 Foreignkey

1.建立資料表及外來鍵關係

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

from sqlalchemy import Column,Integer,String,ForeignKey
from sqlalchemy.orm import relationship



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

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

    # 關聯欄位,讓class_id 與 class 的 id 進行關聯,主外來鍵關係(這裡的ForeignKey一定要是表名.id不是物件, 這是資料庫層面的
    school_id = Column(Integer,ForeignKey('school.id'))
    
    # orm層面的關係,資料庫中不存在這個欄位
    to_school = relationship("School",backref='to_student')

Base.metadata.create_all(engine)

新增資料

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

from sqlalchemy import Column,Integer,String,ForeignKey
from sqlalchemy.orm import relationship



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

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

    # 關聯欄位,讓class_id 與 class 的 id 進行關聯,主外來鍵關係(這裡的ForeignKey一定要是表名.id不是物件
    school_id = Column(Integer,ForeignKey('school.id'))

    to_school = relationship("School",backref='to_student')

# Base.metadata.create_all(engine)

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(engine)
db_session = Session()

# 正向新增欄位
s = Student(name='小明',to_school=School(name='太陽小學'))
s2 = Student(name='小亮',to_school=db_session.query(School).filter(School.name=='太陽小學').first())
db_session.add(s)

# 反向新增欄位
school = School(name='月亮小學')
school.to_student=[Student(name='xxx1'),Student(name='xxx2')]
db_session.add(school)
db_session.commit()

查詢資料

# 正向查詢
student_list = db_session.query(Student).all()
for stu in student_list:
    print(stu.name,stu.to_school.name)

# 反向查詢
school_list = db_session.query(School).all()
for school in school_list:
    print(school.name,end='')
    for s in school.to_student:
        print(s.name)

修改資料

school = db_session.query(School).filter(School.name=='太陽小學').first()
db_session.query(Student).filter(Student.school_id==school.id).update({'name':'太陽小學生'})

db_session.commit()

刪除資料

school = db_session.query(School).filter(School.name=='太陽小學').first()
db_session.query(Student).filter(Student.id==school.id).delete()
 
db_session.commit()
db_session.close()

多對多關係

建立表及關係

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test',encoding='utf8')

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

from sqlalchemy import Integer,Column,String,ForeignKey
from sqlalchemy.orm import relationship

class Boy(Base):
    __tablename__ = 'boy'
    id = Column(Integer,primary_key=True)
    name = Column(String(16))

    to_girl = relationship('Boy',secondary='hotel',backref='to_boy')

class Girl(Base):
    __tablename__ = 'girl'
    id = Column(Integer,primary_key=True)
    name = Column(String(16))

class Hotel(Base):
    __tablename__ = 'hotel'
    id = Column(Integer,primary_key=True)
    boy_id = Column(Integer,ForeignKey('boy.id'))
    girl_id = Column(Integer,ForeignKey('girl.id'))

Base.metadata.create_all(engine)

基於relationship增加資料 多對多

boy = Boy(name='男孩1')
boy.b_to_g = [Girl(name='女1'),Girl(name='女2')]

db_session.add(boy)
db_session.commit()

基於relationship查詢資料 多對多

# 建立連線
from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]/test?charset=utf8')

from sqlalchemy.ext.declarative import declarative_base
# 建立基類
Base = declarative_base()

from sqlalchemy import Integer,Column,String,ForeignKey
from sqlalchemy.orm import relationship
# 建立表關係
class Boy(Base):
    __tablename__ = 'boy'
    id = Column(Integer,primary_key=True)
    name = Column(String(16))

    b_to_g = relationship('Girl',secondary='hotel',backref='g_to_b')

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

    def __repr__(self):
        return self.name

class Hotel(Base):
    __tablename__ = 'hotel'
    id = Column(Integer,primary_key=True)
    boy_id = Column(Integer,ForeignKey('boy.id'))
    girl_id = Column(Integer,ForeignKey('girl.id'))

# Base.metadata.create_all(engine)

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(engine)
db_session = Session()

# 正向查詢
res = db_session.query(Boy).all()
for boy in res:
    print(boy.id,boy.name,boy.b_to_g)

# 反向查詢
res = db_session.query(Girl).first()
for boy in res.g_to_b:
    print(boy.id,boy.name)