目錄

 

定義一個類，ORM(物件關係對映)將這個類轉換為sql語句，使用pymysql進行執行

一，底層處理

使用engine/connectionpooling/dialect進行資料庫操作，engine使用connectionpooling連線資料庫，然後在通過dialect執行sql語句（SQLAlchemy本身無法操作資料庫，其必須以來pymsql等第三方外掛，Dialect用於和資料API進行交流，根據配置檔案的不同調用不同的資料庫API）

pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]

from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://root:[email protected]:3306/t1",max_overflow=5)

二，建立一個單表

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

engine = create_engine("mysql+pymysql://root:
 
[email protected]:3306/t1")

Base = declarative_base()#生成一個SQLORM基類

class User(Base):
    __tablename__ = "users"

    id = Column(Integer,primary_key=True,autoincrement=True)
    name = Column(String(40))
    fullname = Column(String(40))
    password = Column(String(40))

    def __repr__(self):
        return "<User(name='%s', fullname='%s',password='%s')>"%(self.name,self.fullname,self.password)

Base.metadata.create_all(engine)#建立所有表結構
#Base.metadata.drop_all(engine)#刪除所有的表

ed_user = User(name='xiaoyu',fullname='xiaoudaf',password='123')

MySession = sessionmaker(bind=engine)
session=MySession()

#一次加入一條
# session.add(ed_user)

#一次加入多條
# session.add_all([
#     User(name='daf',fullname='fafwaf',password='fwa'),
#     User(name='daf', fullname='fafwaf', password='fwa'),
#     User(name='daf', fullname='fafwaf', password='fwa'),
# ])
#
# session.commit()

#查詢所有
print(session.query(User).all())
#按順序查詢
for row in session.query(User).order_by(User.id):
    print(row)

#進行過濾
for row in session.query(User).filter(User.name.in_(['xiaoyu',])):
    print(row)

for row in session.query(User).filter(~User.name.in_(['xiaoyu', ])):
    print(row)

for row in session.query(User).filter(User.name=='xiaoyu'):
    print(row)

print(session.query(User).filter(User.name=='xiaoyu').count())


for row in session.query(User).filter(and_(User.name=='daf',User.fullname=='fafwafs')):
    print(row)

for row in session.query(User).filter(or_(User.name=='daf',User.fullname=='xiaoudafc')):
    print(row)