flask异步（flask异步任务）

本篇文章给大家谈谈flask异步，以及flask异步任务对应的知识点，希望对各位有所帮助，不要忘了收藏本站喔。

本文目录一览：

• 1、在python3下怎样用flask-sqlalchemy对mysql数据库操作
• 2、uwsgi和flask部署之后为什么一直提示不能导入flask
• 3、flask配置生产环境
• 4、Python几种主流框架比较
• 5、Flask 实现异步服务

在python3下怎样用flask-sqlalchemy对mysql数据库操作

以 Debian/Ubuntu 为例（请确保有管理员权限）：

1.MySQL

代码如下:

apt-get install mysql-server

apt-get install mysql-client

apt-get install libmysqlclient15-dev

2.python-mysqldb

代码如下:

apt-get install python-mysqldb

3.easy_install

代码首或如下:

wget

python ez_setup.py

4.MySQL-Python

代码如下:

easy_install MySQL-Python

5.SQLAlchemy

代码如下:

easy_install SQLAlchemy

6、安装完成后使用下凳乱面代码测试连接

代码如下：

from sqlalchemy import create_engine

from sqlalchemy.orm import sessionmaker

DB_CONNECT_STRING = 'mysql+mysqldb://root:123@localhost/ooxx?charset=utf8'

engine = create_engine(DB_CONNECT_STRING, echo=True)

DB_Session = sessionmaker(bind=engine)

session = DB_Session()

7、数据操作（增删改查）

代码如下：

from sqlalchemy import func, or_, not_

user = User(name='a')

session.add(user)

user = User(name='b')

session.add(user)

user = User(name='a')

session.add(user)

user = User()

session.add(user)

session.commit()

query = session.query(User)

print query # 显示者粗伍SQL 语句

print query.statement # 同上

for user in query: # 遍历时查询

print user.name

print query.all() # 返回的是一个类似列表的对象

print query.first().name # 记录不存在时，first() 会返回 None

# print query.one().name # 不存在，或有多行记录时会抛出异常

print query.filter(User.id == 2).first().name

print query.get(2).name # 以主键获取，等效于上句

print query.filter('id = 2').first().name # 支持字符串

query2 = session.query(User.name)

print query2.all() # 每行是个元组

print query2.limit(1).all() # 最多返回 1 条记录

print query2.offset(1).all() # 从第 2 条记录开始返回

print query2.order_by(User.name).all()

print query2.order_by('name').all()

print query2.order_by(User.name.desc()).all()

print query2.order_by('name desc').all()

print session.query(User.id).order_by(User.name.desc(), User.id).all()

print query2.filter(User.id == 1).scalar() # 如果有记录，返回第一条记录的第一个元素

print session.query('id').select_from(User).filter('id = 1').scalar()

print query2.filter(User.id 1, User.name != 'a').scalar() # and

query3 = query2.filter(User.id 1) # 多次拼接的 filter 也是 and

query3 = query3.filter(User.name != 'a')

print query3.scalar()

print query2.filter(or_(User.id == 1, User.id == 2)).all() # or

print query2.filter(User.id.in_((1, 2))).all() # in

query4 = session.query(User.id)

print query4.filter(User.name == None).scalar()

print query4.filter('name is null').scalar()

print query4.filter(not_(User.name == None)).all() # not

print query4.filter(User.name != None).all()

print query4.count()

print session.query(func.count('*')).select_from(User).scalar()

print session.query(func.count('1')).select_from(User).scalar()

print session.query(func.count(User.id)).scalar()

print session.query(func.count('*')).filter(User.id 0).scalar() # filter() 中包含 User，因此不需要指定表

print session.query(func.count('*')).filter(User.name == 'a').limit(1).scalar() == 1 # 可以用 limit() 限制 count() 的返回数

print session.query(func.sum(User.id)).scalar()

print session.query(func.now()).scalar() # func 后可以跟任意函数名，只要该数据库支持

print session.query(func.current_timestamp()).scalar()

print session.query(func.md5(User.name)).filter(User.id == 1).scalar()

query.filter(User.id == 1).update({User.name: 'c'})

user = query.get(1)

print user.name

user.name = 'd'

session.flush() # 写数据库，但并不提交

print query.get(1).name

session.delete(user)

session.flush()

print query.get(1)

session.rollback()

print query.get(1).name

query.filter(User.id == 1).delete()

session.commit()

print query.get(1)

uwsgi和flask部署之后为什么一直提示不能导入flask

直接使用 gunicorn 就挺好的，uWSGI 安装配置太麻渗帆渗烦了，而且它的异步模式在 Flask 上可能出现问题。

创建目录（轿滚site_doc是main.py的存放目录，/home/python_venv是你的python虚拟环境）

mkdir -p /home/python_venv \

mkdir -p /home/site_doc \

chown -R www:www /site_doc \

chmod -R 775 /site_doc

# 安装 virtualenv ###########################

easy_install virtualenv \

virtualenv --no-site-packages /home/python_venv

# 安装 xx 到虚拟环境 ###########################

# 我丛脊安装了flask，你的bottle就换成bottle

source /home/python_venv/bin/activate \

easy_install Flask \

deactivate

# nginx配置

location / {

uwsgi_pass 127.0.0.1:8888;

include uwsgi_params;

uwsgi_param UWSGI_PYHOME /home/python_venv;

uwsgi_param UWSGI_CHDIR /home/site_doc;

uwsgi_param UWSGI_SCRIPT main;

}

# uwsgi启动

/usr/local/uwsgi/uwsgi -s :8888 -M -p 4 -t 30 -R 10000 --post-buffering 40960 -d /usr/local/uwsgi/logs/uwsgi.log --disable-logging --close-on-exec --env PYTHON_EGG_CACHE=/tmp --socket-timeout 60 --http-timeout 60 --limit-as 512 --gid www --uid www --no-site --vhost

[img]

flask配置生产环境

书接上文 flask与SQLAlchemy ，我们了解到

这一步的的测试可以看 一个小型的资源管理站

这里与直接上结果。

原flask使用的WSGI是python，但是很明显python作为HTTP解释器是效率低下的。而并耐清且，直接运行flask也是非常糟心的。比如集中配置，比如多进程多线程，比如socket通信，比如log管理，比如进程管理

将此文件放置在项目的根目录，并创建文件夹uwsgi

详见

flask+uWSGI已经能实现webserver的功能了，但是更加优秀的weberver还需要nginx的辅助。

Nginx（发音同engine x）是异步框架的网页服务器，也可以用作反向代理、负载平亩陵衡器和HTTP缓存。

还有写拓展处理 主要看服务配置

这里 nginx -s reload 已经完成了服务配置。如果不想配置服务，直接到nginx根目录运行 ./sbin/nginx

这里给绝前大家推荐一个不错的平台：Cloudflare

Cloudflare

在使用之前我们需要更换自己的顶级域名代理商，这一步此处以腾讯云截图：

跳过注册等过程，我们直接进入域名管理界面，看到

然后创建密钥

xxx.pem

xxx.key

再扔到服务器上去。

最后再配置nginx

总结配置flask一般形态的生产环境nginx+uwsgi+flask

回到目录

Python几种主流框架比较

从GitHub中整理出的15个最受欢迎的Python开源框架。这些框架包括事件I/O，OLAP，Web开发，高性能网络通信，测试，爬虫等。

Django: Python Web应用开发框架

Django 应该是最出名的Python框架，GAE甚至Erlang都有框架受它影响。Django是走大而全的方向，它最出名的是其全自动化的管理后台：只需要使用起ORM，做简单的对象定义，它就能自动生成数据库结构、以及全功能的管理后台。

Diesel：基于Greenlet的事件I/O框架

Diesel提供一个整洁的API来编写网络客户端和服务器。支持TCP和UDP。

Flask：一个用Python编写的轻量级Web应用框架

Flask是一个使用Python编写的轻量级Web应用框架。基于Werkzeug WSGI工具箱和Jinja2

模板引擎。Flask也被称为“microframework”，因为它使用简单的核心，用extension增加其他功能。Flask没有默认使用的数

据库、窗体验证工具源租。

Cubes：轻量级Python OLAP框架

Cubes是一个轻量级Python框架，包含OLAP、多维数据分析和浏览聚合数据（aggregated data）等工具。

Kartograph.py：创造矢量地图的轻量级Python框架

Kartograph是一个Python库，用来为ESRI生成SVG地图。Kartograph.py目前仍处于beta阶段，你可以在virtualenv环境下来测试。

Pulsar：Python的事件驱动并发框架

Pulsar是一个事件驱动的并发框架，有了pulsar，你可以写出在不同进程或线程中运行一个或多个活动的异步服务器。

Web2py：全栈式Web框架

Web2py是一个为Python语言提供的全功能Web应用框架，旨在敏捷快速的开发Web应用，具有快速、安全以及可移植的数据库驱动的应用，兼容Google App Engine。

Falcon：构建云API和网络应用后端的高性能Python框架

Falcon是一个构建云API的高性能Python框架，它鼓励使用REST架构风格，尽可能以最少的力气做最多的事情。

Dpark：Python版的Spark

DPark是Spark的Python克隆，是一个Python实现的分布式计算框架，可以非常方便地实现大规模数据处理和迭代计算。DPark由豆瓣实现，目前豆瓣内部的绝大多数数据分析都使用DPark完成，正日趋完善。

Buildbot：基于Python的持续集成测试框架

Buildbot是一个开源框架，可以自动化软件构建、测试和发布等过程。每当代码有改变，服务器要求不同平台上的客户端立即进行代码构建和测试，收集并报告不同平台的构建和测试结果。

Zerorpc：基于ZeroMQ的高性能分布式RPC框架

Zerorpc是一个基于ZeroMQ和MessagePack开发的远程过程调用协议（RPC）实现。和 Zerorpc 一起使用的 Service API 被称为 zeroservice。Zerorpc 可以通过编程或命令镇手行方式调用。

Bottle: 微型Python Web框架

Bottle是一个简单高效的遵循WSGI的御裂嫌微型python Web框架。说微型，是因为它只有一个文件，除Python标准库外，它不依赖于任何第三方模块。

Tornado：异步非阻塞IO的Python Web框架

Tornado的全称是Torado Web Server，从名字上看就可知道它可以用作Web服务器，但同时它也是一个Python Web的开发框架。最初是在FriendFeed公司的网站上使用，FaceBook收购了之后便开源了出来。

webpy: 轻量级的Python Web框架

webpy的设计理念力求精简（Keep it simple and powerful），源码很简短，只提供一个框架所必须的东西，不依赖大量的第三方模块，它没有URL路由、没有模板也没有数据库的访问。

Scrapy：Python的爬虫框架

Scrapy是一个使用Python编写的，轻量级的，简单轻巧，并且使用起来非常的方便。

Flask 实现异步服务

Flask的服务，默认是同步的，在接收多个请求是会发生阻塞的，导致打开页面变的很慢，很卡，如下一个flask服务：

# -*- coding: utf-8 -*-

import sys

from seleniumimport webdriver

from  bs4import BeautifulSoup

import time

import json

from pymongoimport MongoClient

from zhimaipimport getdailione

from flaskimport Flask, jsonify, request

#创建一个Flask对象

app = Flask(__name__)

def browserini():

global driver

# 启动浏览器

# ip, exttime = getdailione(3)  #调用芝麻IP 3-6小时

    ip ='47.96.225.239:4111'

    while ip ==0:

# ip, exttime = getdailione(3)

        ip ='47.96.225.239:4111'

        time.sleep(3)

chrome_options = webdriver.ChromeOptions()

chrome_options .add_argument('--headless')

chrome_options.add_argument('--disable-gpu')

chrome_options.add_argument('--proxy-server=' + ip)

driver = webdriver.Chrome(chrome_options=chrome_options)# 加载浏览器汪乱搭驱动

# driver.set_window_size(1920, 1080)

 困拿   url =''

#flask服务，供单点查询调用

@app.route('/api/search/baseinfo/',methods=['POST'])

def baseinfo():

url =''

    driver.get(url)

search_name = request.form.get('searchname')# 获取表单请求参数

    if(len(search_name)5):

return json.dumps({'error_message':'sorry, please enter the correct company name!'})

else:

print ('query company_name is:',search_name)

element = driver.find_element_by_id('searchkey')#搜索框

 陪大       time.sleep(3)

element.send_keys(search_name)#输入搜索公司名称

        clickbutton = driver.find_element_by_id('V3_Search_bt')#搜索按钮

        clickbutton.click()

#进入搜索公司列表，点击公司链接

        soup = BeautifulSoup(driver.page_source,'lxml')

num = soup.find('span', {'id':'countOld'}).find('span', {'class':'text-danger'}).text.strip()# 搜寻到该公司数量列表

        if  '0'==num:

print u'抱歉！查询不到该公司信息,请确认后重新获取!'

            return json.dumps({'message':u'抱歉！查询不到该公司信息,请确认后重新获取!'})

else:

hrefbutton = driver.find_element_by_class_name('ma_h1')#找到第一个公司链接点击

            hrefbutton.click()

handles = driver.window_handles#当前所有句柄

            print handles

first_handles = handles[0]

driver.switch_to.window(handles[1])#将浏览器驱动跳转到当前窗口

#开始解析页面获取基本信息

            soup = BeautifulSoup(driver.page_source)

company_name = soup.find('div', {'id':'company-top'}).find('div',{'class' :'content'}).find('div', {'class':'row title'}).text.strip()#公司名称

#print company_name, type(company_name)

            company_name = company_name.split(' ')[0].split('\n')[0]

#基础信息标签

            baselist = soup.find('section', {'id':'Cominfo'}).find_all('table', {'class':'ntable'})[1]#

            tr_list = baselist.find_all('tr')

# 插入基本信息数据表

            data= json.dumps({

# 注册资本

                'register_capital': tr_list[0].find_all('td')[1].text.strip(),

# 实缴资本

                'real_capital': tr_list[0].find_all('td')[3].text.strip(),

# 经营状态

                'operate_state': tr_list[1].find_all('td')[1].text.strip(),

# 公司名称     

'company_name' : company_name,

# 爬取时间

                'crawl_time': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))

})

# 关闭新窗口

    driver.close()

driver.switch_to.window(first_handles)# 将浏览器驱动跳转到当前窗口

    return data

if __name__ =='__main__':

try:

# 启动浏览器

# ip, exttime = getdailione(3)  #调用芝麻IP 3-6小时

        ip ='47.96.225.239:4111'

        while ip ==0:

ip, exttime = getdailione(1)

#ip = '47.96.225.239:4111'

            time.sleep(3)

chrome_options = webdriver.ChromeOptions()

chrome_options .add_argument('--headless')

chrome_options.add_argument('--disable-gpu')

chrome_options.add_argument('--proxy-server=' + ip)

driver = webdriver.Chrome(chrome_options=chrome_options)# 加载浏览器驱动

#driver.set_window_size(1920, 1080)

        time.sleep(5)

except:

browserini()

app.run(host='0.0.0.0',port =5008,debug=True,use_reloader=False)

比较简单的实现异步的方式就是借助第三方库：gevent

# -*- coding: utf-8 -*-

from gevent import monkey

from gevent import pywsgi 或者 from gevent.pywsgi import WSGIServer

from flask import Flask

import requests

app = Flask(__name__)

@app.route('/')

def index():

        #具体的处理逻辑

http_server = pywsgi.WSGIServer(('127.0.0.1', 5000), app)

http_server.serve_forever()

关于flask异步和flask异步任务的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。