flask-sqlalchemy組件

時間  2019-11-17
標籤 flask sqlalchemy 組件 欄目 Python 简体版
原文   原文鏈接

1、簡介

　　flask自己沒有內置orm框架，須要依賴第三方模塊，這裏介紹flask-sqlalchemy,而flask-sqlalchemy是一個flask的擴展，本質上是對sqlalchemy的進一步封裝，固然也須要sqlalchemy的支持，使用起來和原本的sqlalchemy的orm是同樣的。本文主要介紹sqlalchemy的使用以及如何在flask中是用orm。 

2、sqlalchemy

　　SQLAlchemy是Python編程語言下的一款ORM框架，該框架創建在數據庫API之上，使用關係對象映射進行數據庫操做，簡言之即是：將對象轉換成SQL，而後使用數據API執行SQL並獲取執行結果。圖示：html

基本使用

安裝：python

pip3 install sqlalchemy

注：SQLAlchemy沒法修改表結構，若是須要可使用SQLAlchemy開發者開源的另一個軟件Alembic來完成，官網doc：http://docs.sqlalchemy.org/en/latest/core/expression_api.htmlmysql

原生SQLlinux

使用 Engine/ConnectionPooling/Dialect 進行數據庫操做，Engine使用ConnectionPooling鏈接數據庫，而後再經過Dialect執行SQL語句redis

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#Author:wd
from sqlalchemy import create_engine
engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student", max_overflow=5)#建立鏈接，容許溢出5個鏈接
result = engine.execute('select * from student')#使用excute執行原生sql
print(result.fetchall())#獲取全部結果，與pymyql相似

事務sql

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#Author:wd
from sqlalchemy import create_engine
engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student", max_overflow=5)#建立鏈接，容許溢出5個鏈接
result = engine.execute('select * from student')#使用excute執行原生sql
with engine.begin() as conn: #事務操做
    conn.execute("insert into student (name, age, res_date) values ('weikang', 33, '1992-11-11')")
    
print(result.fetchall())#獲取全部結果，與pymyql相似

建表

　　定義數據表，才能進行sql表達式的操做，畢竟sql表達式的表的肯定，是sqlalchemy制定的，若是數據庫已經存在了數據表還須要定義麼？固然，這裏實際上是一個映射關係，若是不指定，查詢表達式就不知道是附加在那個表的操做，固然定義的時候，注意表名和字段名，代碼和數據的必須保持一致。定義好以後,就能建立數據表，一旦建立了，再次運行建立的代碼，數據庫是不會建立的。數據庫

　　sqlalchemy內部組件調用順序爲：使用 Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 進行數據庫操做。Engine使用Schema Type建立一個特定的結構對象，以後經過SQL Expression Language將該對象轉換成SQL語句，而後經過 ConnectionPooling 鏈接數據庫，再而後經過 Dialect 執行SQL，並獲取結果。express

TIPS：使用類的方式和使用metadata方式建立表時候區別在於metadata能夠不指定主鍵，而是用class方式必需要求有主鍵。django

demo1:編程

from sqlalchemy import create_engine,Table,Column,Integer,String,ForeignKey,MetaData

engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     encoding="utf-8",
                     echo=True,
                      max_overflow=5
                     )
#?charset是字符集編碼，echo=True打印輸出信息和執行的sql語句默認Flase，max_overflow=5容許溢出鏈接池鏈接數量
meta=MetaData()#生成源類
#定義表結構
user=Table('user',meta,
           Column('id',Integer,nullable=Table,autoincrement=True,primary_key=True),
           Column('name',String(20),nullable=True),
           Column('age',Integer,nullable=True)
           )

host=Table('host',meta,
            Column('ip',String(20),nullable=True),
            Column('hostname',String(20),nullable=True),

)
meta.create_all(engine)#建立表，若是存在則忽視

demo2:

使用orm基類建立

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#Author:wd
from sqlalchemy import create_engine,Table,Column,Integer,String,ForeignKey,MetaData,Date
from sqlalchemy.ext.declarative import declarative_base

engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     encoding="utf-8",
                     echo=True,
                      max_overflow=5
                     )
#?charset是字符集編碼，echo=True打印輸出信息和執行的sql語句默認Flase，max_overflow=5容許溢出鏈接池鏈接數量
base=declarative_base()#生成ORM基類
#定義表結構
class User(base):
    __tablename__='book' #代表
    id = Column(Integer, primary_key=True)
    name=Column(String(32))
    date=Column(Date)

base.metadata.create_all(engine)#建立表，若是存在則忽視

orm增刪改查

增

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#Author:wd
from sqlalchemy import create_engine,Table,Column,Integer,String,ForeignKey
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     max_overflow=5,
                     echo=True)
#數據庫鏈接信息爲，鏈接類型://用戶名:密碼@數據庫地址:端口／數據庫名字？編碼
#max_overflow建立鏈接，容許溢出5個鏈接,echo=True，輸出相應的sql信息到控制檯，方便調試。

base=declarative_base()#生成orm基類

class user(base):  #映射表
    __tablename__='user'
    id=Column(Integer,autoincrement=True,primary_key=True)
    name=Column(String(20))
    age=Column(Integer)

sessoion_class=sessionmaker(bind=engine)#建立與數據庫的會話類，這裏的sessoion_class是類
Session=sessoion_class()#生成會話實例
user1=user(name='wd',age=22)#生成user對象
Session.add(user1)   #添加user1，可使用add_all,參數爲列表或者tuple
Session.commit()     #提交
#Session.rollback()  #回滾
Session.close()      #關閉會話

刪

data=Session.query(user).filter(user.age==33).delete()
Session.commit()        #提交
Session.close()      #關閉會話

改

#data=Session.query(user).filter(user.age>20).update({"name":'jarry'})#update語法
data=Session.query(user).filter(user.age==22).first()#面向對象語法
data.name='coco'#若是data中數據多條須要使用for循環設置
Session.commit()        #提交
Session.close()      #關閉會話

查

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#Author:wd
from sqlalchemy import create_engine,Table,Column,Integer,String,ForeignKey
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     max_overflow=5,
                     echo=True)
#數據庫鏈接信息爲，鏈接類型://用戶名:密碼@數據庫地址:端口／數據庫名字？編碼
#max_overflow建立鏈接，容許溢出5個鏈接,echo=True，輸出相應的sql信息到控制檯，方便調試。

base=declarative_base()#生成orm基類

class user(base):  #映射表
    __tablename__='user'
    id=Column(Integer,autoincrement=True,primary_key=True)
    name=Column(String(20))
    age=Column(Integer)

    def __repr__(self):   #定義
        return "(%s,%s,%s)" % (self.id,self.name,self.age)

sessoion_class=sessionmaker(bind=engine)#建立與數據庫的會話類，這裏的sessoion_class是類
Session=sessoion_class()#生成會話實例



#data=Session.query(user).get(2)  #get語法獲取primrykey中的關鍵字，在這裏主鍵爲id，獲取id爲2的數據
#data=Session.query(user).filter(user.age>22,user.name=='mack').first()
#filter語法兩個等於號，filter_by語法一個等於號，能夠有多個filter,若是多個數據返回列表，first表明獲取第一個，爲all()獲取全部
data=Session.query(user).filter(user.age>20,user.name.in_(['mack','wd'])).all()#in語法
print(data[0]) #打印第一個結果
Session.commit()        #提交，若是回滾的話，數據將不存在了
Session.close()      #關閉會話
Common Filter Operators

Here’s a rundown of some of the most common operators used in filter():

equals:

     query.filter(User.name == 'ed')
not equals:

     query.filter(User.name != 'ed')
LIKE:

query.filter(User.name.like('%ed%'))

IN:

NOT IN:
query.filter(~User.name.in_(['ed', 'wendy', 'jack']))

IS NULL:

IS NOT NULL:

AND:
2.1. ObjectRelationalTutorial 17

query.filter(User.name.in_(['ed', 'wendy', 'jack']))
# works with query objects too:
query.filter(User.name.in_( session.query(User.name).filter(User.name.like('%ed%'))

))

query.filter(User.name == None)
# alternatively, if pep8/linters are a concern
query.filter(User.name.is_(None))
query.filter(User.name != None)
# alternatively, if pep8/linters are a concern
query.filter(User.name.isnot(None))
SQLAlchemy Documentation, Release 1.1.0b1

# use and_()

from sqlalchemy import and_
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))

# or send multiple expressions to .filter()
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
# or chain multiple filter()/filter_by() calls
query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')
Note: Makesureyouuseand_()andnotthePythonandoperator! • OR:

Note: Makesureyouuseor_()andnotthePythonoroperator! • MATCH:

query.filter(User.name.match('wendy'))
Note: match() uses a database-specific MATCH or CONTAINS f 

經常使用查詢語法
經常使用查詢

其餘操做

##獲取全部數據
data=Session.query(user).all()#獲取user表全部數據
for i in data:
    print(i)


##統計
#count=Session.query(user).count()#獲取全部的條數
count=Session.query(user).filter(user.name.like("ja%")).count()#獲取某些條數
print(count)


##分組
from sqlalchemy import func#須要導入func函數
res=Session.query(func.count(user.name),user.name).group_by(user.name).all()
print(res)

外間關聯

TIPS：設置外檢的另外一種方式 ForeignKeyConstraint(['other_id'], ['othertable.other_id'])

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#Author:wd
from sqlalchemy import create_engine,Table,Column,Integer,String,ForeignKey
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     encoding="utf-8",
                     echo=True,
                      max_overflow=5
                     )
#?charset是鏈接數據庫的字符集編碼（和數據庫的編碼同樣），echo=True打印輸出信息和執行的sql語句默認Flase，max_overflow=5容許溢出鏈接池鏈接數量


Base=declarative_base()

class user(Base):
    __tablename__='user'
    id=Column(Integer,primary_key=True,autoincrement=True)
    name=Column(String(20))
    age=Column(Integer)
    def __repr__(self):
        return "<id:%s,name:%s,age:%s>"%(self.id,self.name,self.age)

class host(Base):
    __tablename__='host'
    user_id=Column(Integer,ForeignKey('user.id'))#user_id關聯user表中的id
    hostname=Column(String(20))
    ip=Column(String(20),primary_key=True)
    host_user=relationship('user',backref='user_host')
    #經過host_user查詢host表中關聯的user信息,經過user_host，在user表查詢關聯的host，與生成的表結構無關，只是爲了方便查詢
    def __repr__(self):
        return "<user_id:%s,hostname:%s,ip:%s>"%(self.user_id,self.hostname,self.ip)

Base.metadata.create_all(engine)
Session_class=sessionmaker(bind=engine)
Session=Session_class()
host1=Session.query(host).first()
print(host1.host_user)
print(host1)
user1=Session.query(user).first()
print(user1.user_host)

多外鍵關聯一個表中的一個字段

應用場景：當咱們購物時候，你會發現有一個收發票地址，和一個收貨地址。關係以下：默認狀況下，發票地址和收穫地址是一致的，可是也有可能我想買東西送給別人，而發票要本身留着，那收貨的地址和寄送發票的地址能夠不一樣。即：同一我的的兩個收穫地址能夠不一樣，多個收穫地址關聯同一我的。

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#Author:wd
from sqlalchemy import Integer, ForeignKey, String, Column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()


class Customer(Base):
    __tablename__ = 'customer'
    id = Column(Integer, primary_key=True)
    name = Column(String)

    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])
    #同時關聯同一個字段，使用relationship須要指定foreign_keys，爲了讓sqlalchemy清楚關聯的外鍵

class Address(Base):
    __tablename__ = 'address'
    id = Column(Integer, primary_key=True)
    street = Column(String)
    city = Column(String)
    state = Column(String)

多對多外鍵關聯

不少時候，咱們會使用多對多外鍵關聯，例如：書和做者，學生和課程，即：書能夠有多個做者，而每一個做者能夠寫多本書，orm提供了更簡單方式操做多對多關係，在進行刪除操做的時候，orm會自動刪除相關聯的數據。

表結構建立

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#Author:wd
from sqlalchemy import Column,Table,String,Integer,ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import relationship

engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     encoding="utf-8",
                     echo=True,
                      max_overflow=5
                     )

Base=declarative_base()
stu_cour=Table('stu_cour',Base.metadata,
               Column('stu_id',Integer,ForeignKey('student.id')),
               Column('cour_id',Integer,ForeignKey('course.id'))
               )

class student(Base):
    __tablename__='student'
    id=Column(Integer,autoincrement=True,primary_key=True)
    stu_name=Column(String(32))
    stu_age=Column(String(32))
    courses=relationship('course',secondary=stu_cour,backref='students')
    #course是關聯的第一張表，stu_cour是關聯的第二張表，固然，也能夠在第三張關聯表中使用兩個relationship關聯student表和course表
    def __repr__(self):
        return '<%s>'%self.stu_name


class course(Base):
    __tablename__='course'
    id=Column(Integer,autoincrement=True,primary_key=True)
    cour_name=Column(String(32))
    def __repr__(self):
        return '<%s>'%self.cour_name
Base.metadata.create_all(engine)

建立表結構
建表

插入數據

#!/usr/bin/env python3
#_*_ coding:utf-8 _*_
#Author:wd
from sqlalchemy import Column,Table,String,Integer,ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import relationship
from sqlalchemy.orm import sessionmaker

engine=create_engine("mysql+pymysql://stu:1234qwer@10.0.0.241:3307/student?charset=gbk",
                     encoding="utf-8",
                     echo=True,
                      max_overflow=5
                     )

Base=declarative_base()
stu_cour=Table('stu_cour',Base.metadata,
               Column('stu_id',Integer,ForeignKey('student.id')),
               Column('cour_id',Integer,ForeignKey('course.id'))
               )

class student(Base):
    __tablename__='student'
    id=Column(Integer,autoincrement=True,primary_key=True)
    stu_name=Column(String(32))
    stu_age=Column(String(32))
    courses=relationship('course',secondary=stu_cour,backref='students')
    #course是關聯的第一張表，stu_cour是關聯的第二張表，固然，也能夠在第三張關聯表中使用兩個relationship關聯student表和course表
    def __repr__(self):
        return '<%s>'%self.stu_name


class course(Base):
    __tablename__='course'
    id=Column(Integer,autoincrement=True,primary_key=True)
    cour_name=Column(String(32))
    def __repr__(self):
        return '<%s>'%self.cour_name

stu1=student(stu_name='wd',stu_age='22')
stu2=student(stu_name='jack',stu_age=33)
stu3=student(stu_name='rose',stu_age=18)
c1=course(cour_name='linux')
c2=course(cour_name='python')
c3=course(cour_name='go')
stu1.courses=[c1,c2]  #添加學生課程關聯
stu2.courses=[c1]
stu3.courses=[c1,c2,c3]
session_class=sessionmaker(bind=engine)
session=session_class()
session.add_all([stu1,stu2,stu3,c1,c2,c3])
session.commit()

數據插入
插入數據

查詢

session_class=sessionmaker(bind=engine)
session=session_class()
stu_obj=session.query(student).filter(student.stu_name=='wd').first()
print(stu_obj.courses)#查詢wd學生所報名的課程
cour_obj=session.query(course).filter(course.cour_name=='python').first()
print(cour_obj.students)#查詢報名python課程所對應的課程
session.commit()
查詢

刪除

session_class=sessionmaker(bind=engine)
session=session_class()
cour_obj=session.query(course).filter(course.cour_name=='python').first()
session.delete(cour_obj)#刪除python課程
session.commit()
刪除

3、flask-sqlalchemy

flask中使用sqlalchemy時候，我的比較習慣安裝django的項目來進行構建，如下示例以登錄驗證進行說明,項目結構：

taskmanager
├── app01
│   ├── __init__.py       #初始化文件
│   ├── models.py        #數據模型
│   └── views
│       └── account.py   # 視圖函數
├── create_table.py    # 建表
├── run.py             # 啓動服務器
├── settings.py        #配置文件
├── static             #靜態資源
└── templates          #模版
    └── login.html

安裝：

pip3 install flask-sqlalchemy

數據庫配置

SQLALCHEMY_DATABASE_URI    #用於鏈接的數據庫 URI 。例如:sqlite:////tmp/test.dbmysql://username:password@server/db
SQLALCHEMY_BINDS     #一個映射 binds 到鏈接 URI 的字典。更多 binds 的信息見用 Binds 操做多個數據庫。
SQLALCHEMY_ECHO    #若是設置爲Ture， SQLAlchemy 會記錄全部 發給 stderr 的語句，這對調試有用。(打印sql語句)
SQLALCHEMY_RECORD_QUERIES    #能夠用於顯式地禁用或啓用查詢記錄。查詢記錄 在調試或測試模式自動啓用。更多信息見get_debug_queries()。
SQLALCHEMY_NATIVE_UNICODE        #能夠用於顯式禁用原生 unicode 支持。當使用 不合適的指定無編碼的數據庫默認值時，這對於 一些數據庫適配器是必須的（好比 Ubuntu 上 某些版本的 PostgreSQL ）。
SQLALCHEMY_POOL_SIZE    #數據庫鏈接池的大小。默認是引擎默認值（一般 是 5 ）
SQLALCHEMY_POOL_TIMEOUT    #設定鏈接池的鏈接超時時間。默認是 10 。
SQLALCHEMY_POOL_RECYCLE    #多少秒後自動回收鏈接。這對 MySQL 是必要的， 它默認移除閒置多於 8 小時的鏈接。注意若是 使用了 MySQL ， Flask-SQLALchemy 自動設定 這個值爲 2 小時。

新創建settings.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# Author:wd
import redis
class BaseConfig(object):
    #session配置
    SESSION_TYPE = 'redis'  # session類型爲redis
    SESSION_KEY_PREFIX = 'session:'  # 保存到session中的值的前綴
    SESSION_PERMANENT = True  # 若是設置爲False，則關閉瀏覽器session就失效。
    SESSION_USE_SIGNER = False  # 是否對發送到瀏覽器上 session:cookie值進行加密
    SESSION_REDIS= redis.Redis(host='10.1.210.33', port='6379')
    #數據庫配置
    SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:1234qwer@10.1.210.33:3306/devops?charset=utf8"
    SQLALCHEMY_POOL_SIZE = 10    #數據庫鏈接池的大小。默認值 5
    SQLALCHEMY_POOL_TIMEOUT = 30  # 指定數據庫鏈接池的超時時間。默認是 10
    SQLALCHEMY_POOL_RECYCLE = -1
    SQLALCHEMY_MAX_OVERFLOW = 3  # 控制在鏈接池達到最大值後能夠建立的鏈接數。當這些額外的鏈接回收到鏈接池後將會被斷開和拋棄
    SQLALCHEMY_TRACK_MODIFICATIONS = False  # 追蹤對象的修改而且發送信號


class ProductionConfig(BaseConfig):
    """生產配置文件"""
    pass


class DevelopmentConfig(BaseConfig):
    """開發配置文件"""
    pass


class TestingConfig(BaseConfig):
    """
    測試配置文件
    """
    pass

在app目錄下創建models.py

from . import db


class UserProfile(db.Model):
    """
    用戶
    """
    __tablename__ = 'userprofile'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, nullable=False)
    password = db.Column(db.String(64), unique=True, nullable=False)
    email = db.Column(db.String(128), unique=True, nullable=False)

    def __repr__(self):
        return '<user %s>' % self.username

建立視圖函數初始化app，App目錄新建__init__.py  
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from .models import *
from .views import account


db = SQLAlchemy()  #實例化

def init_app():
    app = Flask(__name__, template_folder='../templates', static_folder='../static', static_url_path='/static')
    app.config.from_object('settings.DevelopmentConfig')

    # 將db註冊到app中
    db.init_app(app)

    # 註冊藍圖
    app.register_blueprint(account.account)

    return app

在app目錄建立目錄views單個py文件也能夠

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

from flask import Blueprint
from flask import request
from flask import render_template
from .. import db
from .. import models

account = Blueprint('account', __name__)


@account.route('/login',methods=['GET','POST'])
def login():
    if request.method=="GET":
        return render_template("login.html")
    else:
        user_obj=db.session.query(models.UserProfile).filter(models.UserProfile.username==request.form["username"],
                                                             models.UserProfile.password==request.form["password"]).first()
        db.session.close()
        if user_obj:
            return '登錄成功'
        else:
            return render_template("login.html",errors="用戶名或密碼錯誤！")

templates/login.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<div><h1>用戶登錄</h1></div>
<div>
    <form method="post">
        <input type="text" name="username" placeholder="用戶名">
        <input type="password" name="password" placeholder="密碼">
        <input type="submit">
        {{ errors }}
    </form>
</div>
</body>
</html>
login.html

建立啓動腳本run.py  
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# Author:wd

from app01 import init_app

app = init_app()  # 建立app

if __name__ == '__main__':
    app.run()
run.py

離線建立數據庫表腳本create_table.py

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

from app01 import init_app
from app01 import db

app = init_app()  #建立app

with app.app_context():  # 執行腳本建立數據庫表
    db.create_all()
create_table.py
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程