简介
flask是一个轻量级的基于Python的web框架。它没有太多复杂的功能,就像koa,需要一系列的插件来扩展其他功能,被称为microframework。flask没有默认使用的数据库等。
其他Python web框架:Django:它比flask重的多。包含了web开发的常用功能,orm、session、form、admin、分页、中间件、信号、缓存、contenttype等等,Django最著名的是它的全自动化的管理后台:只要使用ORM,做简单的对象定义,就能自动生成数据库结构和管理后台。
Tornado:特性是异步非阻塞,和nodejs很像、原生支持websocket。
什么是虚拟环境一个独立的Python开发环境,不会和系统Python产生干扰
完全可以独立自主运行,包括依赖的安装、卸载等
解决了项目Python运行环境、依赖包版本不统一的问题,不同的Python项目不会产生干扰
目前的两种实现方式virtualenv、pipenv
virtualenvvirtualenv是一个Python安装包
pip install virtualenv
virtualenv --no-site-packages ENV
# cd 到项目下
source ./bin/activate # 激活虚拟环境
decative # 退出环境pipenv
pip install pipenv
pipenv shell # 切换环境
pipenv install package # 安装对应的包到虚拟环境
pipenv lock
pipenv uninstall package
exit # 退出虚拟环境
安装初始化一个独立的运行环境
根据《python环境》的步骤使用pipenv初始化一个独立的运行环境
运行
pipenv shell安装flask
pipenv install flaskMini demo
访问 http://127.0.0.1:3000 , 得到响应数据"hello world":
from flask import Flask
app = Flask(__name__) # 程序名称
@app.route("/", methods=["GET", "POST"]) # 定义路由,指定哪些请求可以响应
def index():
return "hello world"
if __name__ == "__main__": # 在代码不是模块的情况下运行下面的代码
app.run(port=3000, debug=True) # port指定端口;debug为True时,修改代码,程序会自动重启
请求解析GET请求
导入request,从request对象获取请求数据:
浏览器请求:http://127.0.0.1:3000?user=john&age=18
from flask import Flask, request
app = Flask(__name__) # 程序名称
@app.route("/", methods=["GET"]) # 定义路由,指定哪些请求可以响应
def index():
request.args.__str__() # 获取所有的参数
request.args.get("name") # 获取单个参数,输出:john
requets.args.get("age") # 输出 18
return "hello world"
if __name__ == "__main__": # 在代码不是模块的情况下运行下面的代码
app.run(port=3000, debug=True) # port指定端口;debug为True时,修改代码,程序会自动重启POST请求
from flask import Flask
app = Flask(__name__) # 程序名称
@app.route("/", methods=["POST"]) # 定义路由,指定哪些请求可以响应
def index():
request.form # 获取form请求数据
request.form["name"] # 获取form中的name
request.form["age"] # 获取form中的age
request.json # 获取json格式请求数据,会自动将json数据转换为Python的dict或者list
request.json["name"] # 获取json数据中的name
request.json["age"] # 获取json数据中age
return "hello world"
if __name__ == "__main__": # 在代码不是模块的情况下运行下面的代码
app.run(port=3000, debug=True) # port指定端口;debug为True时,修改代码,程序会自动重启
请求响应
使用Response设置响应头
from flask import Flask, Response
import json
class Person:
def __init__(self, name, age):
self.__name__=name
self.__age__=age
def as_dict(self):
return {
"name": self.name,
"age": self.age
}
app = Flask(__name__) # 程序名称
@app.route("/", methods=["POST"]) # 定义路由,指定哪些请求可以响应
def index():
person = Person("John", 18)
return Response(json.dumps(person, default=Person.as_dict), mimetype="application/json") # 使用json.dumps转换为json格式。第一个参数为要转换的数据,第二个参数default为转换为json需要执行的方法。如果第一个参数是list,json.dumps会自动进行遍历。
if __name__ == "__main__": # 在代码不是模块的情况下运行下面的代码
app.run(port=3000, debug=True) # port指定端口;debug为True时,修改代码,程序会自动重启
另外,使用flask中的jsonify函数也可以进行转换。
COOKIE和session
flask中的session借助session实现,COOKIE由Response实现。
from flask import Flask, session, Response
app = Flask(__name__) # 程序名称
@app.route("/login", methods=["GET", "POST"]) # 定义路由,指定哪些请求可以响应
def login():
session["user_name"]=request.json[username] # 取出请求中的用户标识,存储在session中
res = Response("add COOKIE") # 使用COOKIE
res.set_COOKIE(key="COOKIE_demo", value="hello", expires=time.time()+6*60) # 6分钟失效
return "hello world"
@app.route("/logout", methods=["POST"])
def logout():
session.pop("username", None)
return redirect(url_for('login')) # 重定向到login
@app.route("/logout/del_COOKIE", methods=["POST"])
def logout0():
res = Response("del COOKIE") # 删除COOKIE
res.set_COOKIE("COOKIE_demo", "", expires=0) # 删除COOKIE
return res
if __name__ == "__main__": # 在代码不是模块的情况下运行下面的代码
app.run(port=3000, debug=True) # port指定端口;debug为True时,修改代码,程序会自动重启
数据库
flask-sqlalchemy是Falsk的SQLAlchemy的扩展。SQLAlchemy是Python的一个ORM(Object-Relational Mapping)框架。安装
pipenv install flask_sqlalchemy使用
项目结构:
flask初始化 app.py
from flask import Flask
import json
from db import db
from model.url import Url
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://wind:12345678@localhost/wind_dev" # 数据库连接字符串
app.config['SQLALCHEMY_COMMIT_TEARDOWN'] = True
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
db.init_app(app)
@app.route("/")
def index():
return "index page"
@app.route("/urls")
def urls():
url_list = Url.query.all() # 简单查询
return json.dumps(url_list, default=Url.as_dict)
if __name__ == '__main__':
app.run(port=5000, debug=True, threaded=False)
数据库配置 db.py
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
定义model
from db import db
class Url(db.Model):
__tablename__ = 'Urls' # 定义表名
urlId = db.Column(db.String, primary_key=True)
userId = db.Column(db.String)
url = db.Column(db.String)
description = db.Column(db.String)
type = db.Column(db.String)
logo = db.Column(db.String)
createdAt = db.Column(db.DateTime)
updatedAt = db.Column(db.DateTime)
def __init__(self, urlId, userId, url, description, type, logo, createdAt, updatedAt):
self.urlId = urlId
self.userId = userId
self.url = url
self.description = description
self.type = type
self.logo = logo
self.createdAt = createdAt
self.updatedAt = updatedAt
def __repr__(self): # 打印Url的格式,便于调试
return '' % self.url
def as_dict(self): # 转换为json时调用的函数 json.dumps(urls, default=Url.as_dict)
return {
"urlId": self.urlId,
"userId": self.userId,
"url": self.url,
"description": self.description,
"type": self.type,
"logo": self.logo,
"createdAt": self.createdAt.timestamp(),
"updatedAt": self.updatedAt.timestamp()
}
CURD
Url.query.all()
Url.query.filter_by(urlId="1").first()
Url.query.order_by(Url.urlId).all()
Url.query.limit(1).all()
Url.query.get(urlId="1")
url=Url.get(urlId="1")
db.session.delete(url)
db.session.commit()
url=Url("id123", "uid00", "http://www.google.com", "google", "search engine", "http://www.google.com/logo.png", time.timestamp(), time.timestamp())
db.session.add(url)
db.session.commit()