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

上述代码均为测试代码，简单易懂，自行测试即可。

下次给大家讲flask-fom表单空间，涉及到html的知识有前端知识的同学比较易懂。

flask连接数据库mysql+SQLAlchemy

标签：string   hello   mysq   body   sse   定义   none   获取   nullable