程式碼示例

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:
 
[email protected]: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='[email protected]')
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)