Flask 是一個 python web micro framework。所謂微框架，主要是 flask 簡潔與輕巧，自定義程度高。相比 django 更加輕量級。
之前一直折騰 django，得益於django 的 ORM 模式很好用，上手簡單，使用方便。Flask里面沒有原生的 orm，需要用到第三方的庫，
大名頂頂的 SQLALchemy正是一類 實現ORM的庫。
下面簡單介紹一下，Flask中如何使用sqlchemy (注意，這個和 flask-sqlalchemy 的使用還是有差別的，至於 flask-sqlalchemy
用法下一篇博客介紹。)。參考官方文檔 在 Flask 中使用 SQLAlchemy
一 安裝 SQLAlchemy 擴展
在 shell 命令下輸入
pip install sqlchemy
檢測是否安裝成功:
import sqlchemy
sqlchemy.__version__ # 0.10.1
如果沒有報錯，則安裝成功
二 顯式調用
所謂顯示調用，就是免去操作 sqlchemy 和 python class 的映射操作，而是由代碼直接完成。
關於 sqlchemy的詳細使用，再另外一篇博客介紹。移步 sqlchemy的使用教程。
新建一個 flask項目。項目結構如下：
(env)ghost@ghost-H61M-S2V-B3:~/project/flask/fsqlauto$ tree
.
├── app.py
├── database.py
├── models.py
├── static
└── templates
2 directories, 3 files
我們先定義 數據庫連接,操作 database.py
# -*- coding: utf-8 -*-
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///./test.db', convert_unicode=True) # 創建數據庫引擎( 當前目錄下保存數據庫文件)
db_session = scoped_session(sessionmaker(autocommit=False,
autoflush=False,
bind=engine))
Base = declarative_base()
Base.query = db_session.query_property()
def init_db():
# 在這里導入所有的可能與定義模型有關的模塊，這樣他們才會合適地
# 在 metadata 中注冊。否則，您將不得不在第一次執行 init_db() 時
# 先導入他們。
import models
Base.metadata.create_all(bind=engine)
init_db 方法創建數據庫
然后定義我們的 models.py
# -*- coding: utf-8 -*-
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(50), unique=True)
email = Column(String(120), unique=True)
def __init__(self, name=None, email=None):
self.name = name
self.email = email
def __repr__(self):
return '%s (%r, %r)' % (self.__class__.__name__, self.name, self.email)
定義了一個 python class ,實際上是映射了 users 表里.
最后就是我們的應用程序主入口
# -*- coding: utf-8 -*-
from flask import Flask
from database import init_db, db_session
from models import User
app = Flask(__name__)
@app.teardown_request
def shutdown_session(exception=None):
db_session.remove()
@app.route('/')
def index():
return 'hello world flask'
@app.route('/add/<name>/<email>')
def add(name, email):
u = User(name=name, email=email)
try:
db_session.add(u)
db_session.commit()
except Exception, e:
return 'wrong'
return 'Add %s user successfully' % name
@app.route('/get/<name>')
def get(name):
try:
u = User.query.filter(User.name==name).first()
except Exception, e:
return 'there isnot %s' % name
return 'hello %s' % u.name
if __name__ == '__main__':
init_db()
app.debug = True
app.run()
需要注意,定義了一個 @app.teardown_request 裝飾器.用查詢完畢后關閉數據庫,具體可以參考flask文檔 數據庫.
這里的主方法中 init_db 主要是初始化數據庫,如果數據庫存在就鏈接讀取.
add 是數據庫添加記錄方法, get 正好是數據庫查詢,更多例子請參考 sqlchemy教程
三 手動實現 ORM
上面是自動實現 orm,如果是手動,相應的文件修改如下:
database.py
# -*- coding: utf-8 -*-
from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import scoped_session, sessionmaker
engine = create_engine('sqlite:///./test.db', convert_unicode=True)
metadata = MetaData()
db_session = scoped_session(sessionmaker(autocommit=False,
autoflush=False,
bind=engine))
def init_db():
metadata.create_all(bind=engine)
models.py
from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.orm import mapper
from database import metadata, db_session
class User(object):
query = db_session.query_property()
def __init__(self, name=None, email=None):
self.name = name
self.email = email
def __repr__(self):
return '<User %s>' % (self.name)
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50), unique=True),
Column('email', String(120), unique=True)
)
mapper(User, users)
model 需要手動定義 table 並指定映射
app.py
# -*- coding: utf-8 -*-
import sqlite3
from flask import Flask
from database import *
from models import *
app = Flask(__name__)
@app.teardown_request
def teardown_request(exception=None):
db_session.remove()
@app.route('/')
def index():
return 'hello'
@app.route('/add/<name>/<email>')
def add(name, email):
u = User(name, email)
try:
db_session.add(u)
db_session.commit()
except Exception, e:
return 'wrong'
return '%s add successful' % name
@app.route('/get/<name>')
def get(name):
try:
u = User.query.filter(User.name==name).first()
except Exception, e:
return 'there isnot %s' % name
return 'hello %s' % u.name
if __name__ == '__main__':
init_db()
app.debug = True
app.run()
更底層的操作就是 在 flask app 里面寫 sql 查詢，具體和 sqlalchemy 使用方法一致
如果覺得我的文章對您有用，請隨意打賞。您的支持將鼓勵我繼續創作！