SQLAlchemy基本使用，建立表，增刪改查
阿新 • • 發佈：2019-07-02
基礎語法
- 建立連線
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是一個數據庫的名稱，如mysql，oracle， postgresql等等，和driver一個DBAPI的名稱，諸如 psycopg2，pyodbc，cx_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)