简说Python之flask-SQLAlchmey的web应用

目录

系统环境:Ubuntu 18.04.1 LTS

Python使用的是虚拟环境:virutalenv

Python的版本:Python 3.6.9

原生语句操作MySQL数据库

1.安装MySQL

zsd@zsd-virtual-machine:~$ sudo apt-get install mysql-server libmysqlclient-dev -yq

安装PyMySQL[为python3可以驱动MySQL使用]

(zsdpy1) zsd@zsd-virtual-machine:~$ pip install PyMySQL

2.MySQL设置用户和权限

root@zsd-virtual-machine:~# sudo mysql -uroot


mysql> create database zsd;
mysql> create user 'zsd'@'localhost' identified by 'zsd';

mysql> use zsd;
Database changed
mysql> grant all on zsd.* to 'zsd'@'localhost';
Query OK, 0 rows affected (0.00 sec)

3.用PyMySQL操纵MySQL数据库

1.db配置python

(zsdpy1) $ cat dbconfig.py 
# coding=utf-8
HOSTNAME = 'localhost'
DATABASE = 'zsd'
USERNAME = 'zsd'
PASSWORD = 'zsd'
DB_URI = 'mysql+pymysql://{}:{}@{}:3306/{}?charset=utf8'.format(
    USERNAME, PASSWORD, HOSTNAME, DATABASE)
  • HOSTNAME:代表主机名，
  • DATABASE:代表建立的数据库名称
  • USERNAME:进入mysql数据库的用户名
  • PASSWORD:进入mysql的密码
  • DB_URI:用于PyMySQL驱动MySQL数据库的uri。
(zsdpy1) $ cat ex_db01.py 
# coding=utf-8
import pymysql
from dbconfig import HOSTNAME, DATABASE, USERNAME, PASSWORD

try:
    con = pymysql.connect(HOSTNAME, USERNAME, PASSWORD, DATABASE)
    cur = con.cursor()
    cur.execute("SELECT VERSION()")
    ver = cur.fetchone()
    print ("Database version : %s " % ver)
except MySQLdb.Error as e:
    print ("Error %d: %s" % (e.args[0], e.args[1]))
    exit(1)
finally:
    if con:
        con.close()

上述的程序含义：

  • pymysql.connect,建立一个MySQL数据库的连接。
  • con.cursor(),建立一个游标，可以通过SQL语句，操作MySQL数据库的表和数据。
  • cur.execute,执行一条SQL语句。
  • ver = cur.fetchone(),返回第一行数据

可以看到，这里是用于返回数据库的版本数据。

运行效果如下:

(zsdpy1) $  python ex_db01.py 
Database version : 5.7.29-0ubuntu0.18.04.1

4. CRUD增，删，改，查

(zsdpy1) $ cat curd3x.py 
# coding=utf-8
import pymysql
from dbconfig import HOSTNAME, DATABASE, USERNAME, PASSWORD
con = pymysql.connect(HOSTNAME, USERNAME, PASSWORD, DATABASE)

with con as cur:
    cur.execute('drop table if exists users')
    cur.execute('create table users(Id INT PRIMARY KEY AUTO_INCREMENT, '
                'Name VARCHAR(25))')
    cur.execute("insert into users(Name) values('xiaoming')")
    cur.execute("insert into users(Name) values('wanglang')")
    cur.execute('select * from users')

    rows = cur.fetchall()
    for row in rows:
        print (row)
    cur.execute('update users set Name=%s where Id=%s', ('ming', 1))
    print ('Number of rows updated:',  cur.rowcount)

    cur = con.cursor(pymysql.cursors.DictCursor)
    cur.execute('select * from users')

    rows = cur.fetchall()
    for row in rows:
        print (row['Id'], row['Name'])

程序解释:

唯一的区别可以看到通过调用cur.execute,实现你想调用的任何SQL语句，如drop table,insert,update,select,create table等基础SQL语句。

使用SQLAlchemy

1.安装SQLAlchemy

pip install SQLALchemy

各种数据库驱动接口，案例如下:

pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]


MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>


cx_Oracle
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]

实验环境是Python3.6+pymysql,所以用的是第一个。

所以上述的dbconfig.py是一个基础脚本，公用的。

2.原生SQL调用SQLAlchemy

(zsdpy1) $ cat SQLAlchemy_raw_sql3x.py 
# coding=utf-8
from sqlalchemy import create_engine
from dbconfig import DB_URI

eng = create_engine(DB_URI)
with eng.connect() as con:
    con.execute('drop table if exists users')
    con.execute('create table users(Id INT PRIMARY KEY AUTO_INCREMENT, '
                'Name VARCHAR(25))')
    con.execute("insert into users(name) values('zsd')")
    con.execute("insert into users(name) values('lzh')")
    rs = con.execute('select * from users')
    for row in rs:
        print (row)

其中

  • eng = create_engine(DB_URI),建立一个驱动引擎，用于连接MySQL数据库

  • from dbconfig import DB_URI,调用了dbconfig.py的变量:DB_URI 的数据.

调用的结果:

可以看到，我插入的两条数据，zsd和lzh

(zsdpy1)$ python SQLAlchemy_raw_sql3x.py 
(1, 'zsd')
(2, 'lzh')

3.ORM

(zsdpy1) $ cat orm_sql.py 
# coding=utf-8
from sqlalchemy import create_engine, Column, Integer, String, Sequence
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import and_, or_
from sqlalchemy.orm import sessionmaker

from dbconfig import DB_URI

eng = create_engine(DB_URI)
Base = declarative_base()


class User(Base):
    __tablename__ = 'users'
# id是主键，且给与一个Sequence，自增
    id = Column(Integer, Sequence('user_id_seq'),
                primary_key=True, autoincrement=True)
# name是string类型
    name = Column(String(50))
# 删除表，个人实验环境，测试用。
Base.metadata.drop_all(bind=eng)     
Base.metadata.create_all(bind=eng)

# 建立一个session，提供事务控制。
Session = sessionmaker(bind=eng)
session = Session()

# 添加数据，由于id的主键和自增的。所以合理只需添加name的值。
session.add_all([User(name=username)
                 for username in ('zsd', 'lzh', 'huke','ly')])

# 数据库的事务提交，commit
session.commit()

# 获得结果的调用方法
def get_result(rs):
    print ('-' * 20)
# 变量结果集内的name数据
    for user in rs:
        print (user.name)
#查询所有的数据
rs = session.query(User).all()
get_result(rs)
# 相当于select * from user where id =2
rs = session.query(User).filter(User.id.in_([2, ]))
get_result(rs)
# 相当于select * from user where id >2 and id <4
rs = session.query(User).filter(and_(User.id > 2, User.id < 4))
get_result(rs)
# 相当于select * from user where id =2 and id = 4
rs = session.query(User).filter(or_(User.id == 2, User.id == 4))
get_result(rs)
# 相当于select * from user where name like '%s%'
rs = session.query(User).filter(User.name.like('%s%'))
get_result(rs)
# 相当于select * from user where name ='zsd'的第一条数据
user = session.query(User).filter_by(name='ly').first()
get_result([user])

效果如下：

(zsdpy1) $ python orm_sql.py 
--------------------
zsd
lzh
ly
huke
--------------------
lzh
--------------------
ly
--------------------
lzh
huke
--------------------
zsd
--------------------
ly

运用flask-SQLAlchmey

1.安装Flask-SQLAlchemy

pip install Flask-SQLAlchemy

2.flask-SQLAlchmey的web应用

核心的四个文件如下:

(zsdpy1) $ ls -l
total 16
-rw-r--r-- 1 zsd zsd 552 3月  17 16:24 app_with_sqlalchemy.py
-rw-r--r-- 1 zsd zsd 128 3月  17 16:24 config.py
-rw-r--r-- 1 zsd zsd 198 3月  17 16:24 dbconfig.py
-rw-r--r-- 1 zsd zsd 249 3月  17 16:24 users.py
  • dbconfig.py是常用配置文件，上述已经解释过了。

users.py

(zsdpy1) $ cat  users.py 
# coding=utf-8
from ext import db

class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(50))

    def __init__(self, name):
        self.name = name

其中的含义:

  • from ext import db ext存放了Flask第三方的扩展，通过db，传导了db.Model。也体现了ORM中model。
  • id是db模型中的列(Column)，Integer类型.主键，自增id。
  • name是db模型中的列(Column)，name类型。
  • __init__基础语法，用于name的初始化。

config.py

(zsdpy1) zsd@zsd-virtual-machine:~/web_develop/chapter3/section3$ cat config.py 
# coding=utf-8
from dbconfig import DB_URI
DEBUG = True
SQLALCHEMY_DATABASE_URI = DB_URI
SQLALCHEMY_TRACK_MODIFICATIONS = False

继承dbconfig里面的常量，然后可以让flak app去依赖。

app_with_sqlalchemy.py

(zsdpy1) zsd@zsd-virtual-machine:~/web_develop/chapter3/section3$ cat app_with_sqlalchemy.py 
# coding=utf-8
from flask import Flask, request, jsonify

from ext import db
from users import User

app = Flask(__name__)
app.config.from_object('config')
db.init_app(app)

with app.app_context():
    db.drop_all()
    db.create_all()


@app.route('/users', methods=['POST'])
def users():
    username = request.form.get('name')

    user = User(username)

    db.session.add(user)
    db.session.commit()

    return jsonify({'id': user.id})


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=9000)

其中程序的含义:

  • from ext import db 依赖db。
  • app.config.from_object('config')获取config中的常量URI
  • app.app_context()app发生请求之前，做一些操作，这里是删除user表中所有的数据。
  • @app.route('/users', methods=['POST'])以post的方式，获取name参数，后面添加user.nam数据到数据库user表中。

3.实验操作

开启flask app应用

(zsdpy1) $ python app_with_sqlalchemy.py 
 * Serving Flask app "app_with_sqlalchemy" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Running on http://0.0.0.0:9000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 177-062-521

我这里编写了一个http form post的脚本，用于调用这个app应用。

(zsdpy1) zsd@zsd-virtual-machine:~/web_develop/chapter3/section3$ cat http_form.py 
#  -*- coding:utf-8 -*-
import requests
url = "http://0.0.0.0:9000/users"
data = {"name":"zsd2"}
res = requests.post(url=url,data=data)
print(res.text)

开始调用:

(zsdpy1) zsd@zsd-virtual-machine:~/web_develop/chapter3/section3$ python http_form.py    
{
  "id": 1
}

可以看到返回了json串，id=1

查看数据库的数据，如下：

mysql> select * from users;
+----+------+
| id | name |
+----+------+
|  1 | zsd2 |
+----+------+
1 row in set (0.00 sec)

发现，zsd2的数据已经插入，至此，实验结束。enjoy！！！！！

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/zhangshengdong/p/12515795.html