flask连接数据库mysql+SQLAlchemy

# -*- coding: utf-8 -*-
# Flask hello world
from flask import Flask
from flask.ext.mysql import MySQL
app = Flask(__name__)

‘‘‘‘
###链接数据库MySQL版
mysql = MySQL()
app.config[‘MYSQL_DATABASE_USER‘] = ‘root‘
app.config[‘MYSQL_DATABASE_PASSWORD‘] = ‘root‘
app.config[‘MYSQL_DATABASE_DB‘] = ‘test‘
app.config[‘MYSQL_DATABASE_HOST‘] = ‘localhost‘
mysql.init_app(app)
cursor = mysql.connect().cursor()

if __name__ == ‘__main__‘:
    cursor.execute("SELECT * from  db_admin ")
    data = cursor.fetchone()
    print data

‘‘‘
###SQLAlchemy版

__author__ = ‘ghost‘
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey, select, text
# 连接数据库 
engine = create_engine("mysql://root:root@127.0.0.1/test?charset=utf8",encoding="utf-8", echo=True)
# 获取元数据
metadata = MetaData()
# 定义表
user = Table(‘user‘, metadata,
    Column(‘id‘, Integer, primary_key=True),
    Column(‘name‘, String(20)),
    Column(‘fullname‘, String(40)),
  )
 
address = Table(‘address‘, metadata,
    Column(‘id‘, Integer, primary_key=True),
    Column(‘user_id‘, None, ForeignKey(‘user.id‘)),
    Column(‘email‘, String(60), nullable=False)
  )
# 创建数据表，如果数据表存在，则忽视
metadata.create_all(engine)
# 获取数据库连接
conn = engine.connect()
‘‘‘
###插入数据
i = user.insert()
u = dict(name=‘bob‘, fullname=‘bobb‘)
r = conn.execute(i, **u)
####插入多条数据
addresses=[{‘user_id‘: 1, ‘email‘: ‘jack@yahoo.com‘}, {‘user_id‘: 1, ‘email‘: ‘jack@msn.com‘}, {‘user_id‘: 2, ‘email‘: ‘www@www.org‘}, {‘user_id‘: 2, ‘email‘: ‘wendy@aol.com‘}]
a = address.insert()
r = conn.execute(a, addresses)
‘‘‘
####查询多个字段多条数据
s = select([user])
r = conn.execute(s).fetchall()
print r 

####查询多个字段单条数据
s = select([user])
r = conn.execute(s).fetchone()
print r 

###查询单个字段数据
s = select([user.c.id,user.c.name])
r = conn.execute(s).fetchall()
print r

####多表查询
s = select([user.c.name, address.c.email]).where(user.c.id==address.c.user_id)
r = conn.execute(s).fetchall()
print r

####操作链接查询
‘‘‘
se_sql =  [(user.c.fullname +", " + address.c.email).label(‘title‘)]
wh_sql = and_(
            user.c.id == address.c.user_id,
            user.c.name ==‘bob‘,
            or_(
            address.c.email.like(‘%@aol.com‘),
            address.c.email.like(‘%@msn.com‘),
            )
        )
s = select(se_sql).where(wh_sql)
r = conn.execute(s).fetchall()
print r
‘‘‘
#####原生sql
sql = ‘select * from user where id=:id and name=:name‘
s = text(sql)
r = conn.execute(s, id=3, name=‘bob‘).fetchall()
print r

###排序 分组 分页
s = select([user]).order_by(user.c.id)
s = select([user]).order_by(user.c.id.desc())
r = conn.execute(s).fetchall()
print r

s = select([user]).order_by(user.c.id.desc()).limit(3).offset(0)  ### 倒叙取3个
r = conn.execute(s).fetchall()
print r