Flask+SQLAlchemy入门使用

代码示例

1. 插入与查询

# -*- coding: utf-8 -*- from flask import Flask from flask_sqlalchemy import SQLAlchemy from sqlalchemy import create_engine, Table,select, Column, Integer, String, MetaData, ForeignKey app = Flask(__name__) # 配置 sqlalchemy 数据库驱动 app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqlconnector://user:pass@localhost:3306/test?charset=utf8' app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # 初始化 db = SQLAlchemy(app) class User(db.Model): """ 定义了三个字段， 数据库表名为model名小写 """ id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(80), unique=True) fullname = db.Column(db.String(120), unique=True) def __init__(self, name, fullname): self.name = name self.fullname = fullname def __repr__(self): return '<User %r>' % self.username def save(self): db.session.add(self) db.session.commit() u = User(name='admin', fullname='admin@example.com') db.session.add(u) db.session.commit() users = User.query.all()

插入中文需要使用Unicode字符串。

2. 左联接查询

pages = db.session.query(表1, 表2).outerjoin(表2, 表1.SiteId == 表2.SiteId)

3. 条件查询

# conditions if _SiteId: pages = pages.filter(表1.SiteId == _SiteId)

4. 分页

results = results.order_by(sidx + ' ' + sord) pagination = results.paginate(page, per_page=rows, error_out=False) results = pagination.items

5. restful服务

return json.dumps({'d': {'Table': {'rows': new_pages}, 'Total': total, 'ex': '', 'Records': count, 'Page': page}}, cls=AlchemyEncoder)

AlchemyEncoder

# -*- coding: UTF-8 -*- from sqlalchemy.ext.declarative import DeclarativeMeta import json import datetime class AlchemyEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj.__class__, DeclarativeMeta): # an SQLAlchemy class fields = {} for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']: data = obj.__getattribute__(field) try: json.dumps(data) # this will fail on non-encodable values, like other classes fields[field] = data except TypeError: # 添加了对datetime的处理 if isinstance(data, datetime.datetime): fields[field] = data.isoformat() elif isinstance(data, datetime.date): fields[field] = data.isoformat() elif isinstance(data, datetime.timedelta): fields[field] = (datetime.datetime.min + data).time().isoformat() else: fields[field] = None # a json-encodable dict return fields return json.JSONEncoder.default(self, obj)

参考： http://www.jb51.net/article/86018.htm