用werkzeug实现一个简单的python web框架
使用工具
Pycharm , Navicat , WebStorm等
使用库
Werkzeug用于实现框架的底层支撑，pymysql用于实现ORM，jinja2用于模板支持，json用于返回json数据功能的支持
该视图基类用于被视图类所继承，并且提供了两个分别处理GET和POST请求的函数，程序收到请求时，会根据请求的方式将请求参数发送到对应的处理函数中进行处理（请求调度）。若请求方式未找到，则直接返回错误响应。
class View(object):# 请求方式与处理函数对应
def __init__(self):
self.methods = {
'GET': self.GET,
'POST': self.POST
}
# 定义两种基本请求方式
# 视图类可覆盖请求方式进行不同处理
def GET(self, request):
raise MethodNotAllowed()
def POST(self, request):
raise MethodNotAllowed()
# 请求调度
def dispatch_request(self, request, *args, **options):
# 保存request对象至全局变量中
global global_request
global_request = request
# 判断请求类型并将请求分发给相应的处理函数，返回该函数
if request.method in self.methods:
return self.methods[request.method](request, *args, **options)
else:
return '<h1>Unknown or unsupported require method</h1>'
在View类中，使用一个闭包，将视图函数类本身发送给请求调度函数，获取对应的处理函数（对应继承View类的视图类覆盖的请求方法函数）并返回
@classmethoddef get_func(cls):
def func(*args, **kwargs):
obj = func.view_class()
return obj.dispatch_request(*args, **kwargs)
func.view_class = cls
return func
所有视图类必须继承自View类，并至少覆盖其中的GET或POST函数。如
class Index(View):def GET(self,request):
return "hello world"
实现App类
App类拥有两个私有变量，view_func 以及 url_map 分别表示url和视端点的映射字典以及端点和视图函数的映射。依据WSGI_APP的原理，使用实例形式实现App时，必须实现__call__方法，并传入environ字典以及start_response函数，该方法内部根据environ的参数信息，调用特定的处理函数进行处理，并返回封装好的Response对象。
class App(object):def __init__(self):
# url和视端点的映射字典
# 端点和视图函数的映射
self.view_func = {}
self.url_map = Map()
def __call__(self, environ, start_response):
return self.wsgi_app(environ, start_response)
def wsgi_app(self, environ, start_response):
try:
# 构造request对象
request = Request(environ)
# 获取url适配器
adapter = self.url_map.bind_to_environ(request.environ)
# 获取端点值以及动态参数部分
endpoint, values = adapter.match()
# 保存端点值以便获取
request.endpoint = endpoint
# 获取端点对应的视图函数
view = self.view_func.get(endpoint, None)
if view:
# 将请求和url动态部分发送给视图函数获得响应对象
response = view(request, **values)
# 当视图函数返回重定向请求时不再包装成Response对象，直接返回
if response.__class__ != Response:
response = Response(response, content_type='text/html;charset=UTF-8')
else:
response = Response('<h1>404 Not Found<h1>', content_type='text/html; charset=UTF-8')
response.status_code = 404
except HTTPException as e:
response = e
# 返回响应
return response(environ, start_response)
在App类中，我还添加了一个成员方法，用于为该App对象添加路由规则。方法参数为一个字典列表，其中每个字典包括url和view两个键，值分别为视图类对应的路由和视图类。方法内部对该列表进行遍历，并添加到两个成员变量中。
# 添加路由规则def add_url_rule(self, urls):
"""
添加路由规则
:param urls:一个列表，其中每一个项为一个字典，键为url和view，表示路径和对应的视图函数类
:return: None
"""
global url_map
for url in urls:
# 路由url与相应的端点组成键值对
# 默认端点为该视图函数类的类名小写形式
rule = Rule(url["url"], endpoint=url["view"].__name__.lower())
self.url_map.add(rule)
self.view_func[url['view'].__name__.lower()] = url['view'].get_func()
url_map = self.url_map
def run(self, port=5000, ip='127.0.0.1', debug=False):
run_simple(ip, port, self, use_debugger=debug, use_reloader=True)
添加模板支持
可使用jinja2的相关函数为框架添加模板支持
def render_template(template, **ctx):"""
:param template: html文件名
:param ctx: 要传入html 的参数
:return: html标签代码
"""
# 定位template文件夹的路径
global context_processor, global_request
path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "templates")
# 在渲染模板时将一些辅助函数和全局变量传入以便在html代码中使用
ctx["url_for"] = url_for
ctx["request"] = global_request
ctx["session"] = session
ctx["get_flash"] = get_flash
# 将render_template 方法中的键值对参数加入渲染参数列表中
for k, v in context_processor.items():
ctx[k] = v
# 创建jinja环境
jinja_env = Environment(loader=FileSystemLoader(path), autoescape=True)
t = jinja_env.get_template(template)
return t.render(ctx)
使用
from MyFrameWork.myFrame.MyApp import View, create_app, render_templateclass Index(View):
def GET(self,request):
return "hello World"
defurls = [
{
"url":"/index",
"view":Index
}
]app.create_app()
app.add_url_rule(urls)
app.run()
丰富框架功能
我还框架添加了类似flask中的session，flash，url_for，redirect，模板全局变量以及ORM支持等功能。实现方法都比较简单。故不在此赘述。
项目地址（包括框架本身以及使用框架实现的一个小型应用）：https://github.com/YangZX1428/Simple-Python-Web-FrameWork.git
构建url
类似flask的url_for，使用了build方法根据端点构建url，可传入端点和附加参数获取路由路径，也可以访问static中的文件（在添加了支持静态文件的中间件的基础上）
def url_for(endpoint, server_name="127.0.0.1:5000", external=False, filename=None, **values):"""
返回端点值对应的url
:param endpoint: 端点值(会自动转化为小写)
:param server_name: App实例程序所在的服务器ip
:param values: url动态参数部分
:param external: 生成绝对url
:param filename : static资源的路径
:return: 对应的url
"""
# filename不为空时返回静态资源路径
if filename is not None:
file_path = os.path.join('\%s' % endpoint, filename)
return file_path
# 绑定服务器地址
urls = url_map.bind(server_name)
# 通过端点获取对应的url
relative_url = urls.build(endpoint.lower(), values, force_external=external)
return relative_url
返回json格式数据
类似flask的jsonify，可传入键值对，内部将参数转成json数据，并指定响应的content_type为json以返回
def jsonify(**values):"""
返回json格式的响应数据
:param values: 接收键值对
:return: json格式的response
"""
json_data = json.dumps(values)
response = Response(json_data, content_type="application/json;charset=UTF-8")
return response
添加中间件支持
要根据url访问到静态资源，故添加了SharedDataMiddleware中间件
该方法返回一个app对象，相当于创建app
def create_app(with_static=True):"""
创建app对象，加入了中间件
:param with_static:是否开启访问静态资源模式
:return: app对象
"""
app = App()
if with_static:
# 模板中可使用static中的资源
# <link rel=stylesheet href=/static/style.css type=text/css>
app.wsgi_app = SharedDataMiddleware(
app.wsgi_app, {"/static": os.path.join(os.path.dirname(os.path.dirname(__file__)), "static")}
)
return app
重定向
重定向我使用了werkzeug中已经写好的redirect函数，但需要注意的是，redirect函数直接返回Response对象，而在wsgi_app中我们将视图函数的返回值又一次封装成了Response对象，此时当return的是redirect的时候会报错，故在封装时需先判断视图函数的返回值是否已经是Response，若是则直接返回，不是则再包装成Response对象返回
# 将请求和url动态部分发送给视图函数获得响应对象response = view(request, **values)
# 当视图函数返回重定向请求时不再包装成Response对象，直接返回
if response.__class__ != Response:
response = Response(response, content_type='text/html;charset=UTF-8')
ORM支持
我实现的orm目前只支持三种字段类型，String，对应varchar，Integer，对应int，Text，对应text
class StringField(Field):# 字符串字段，字段类型默认为varchar
def __init__(self, name=None, col_type="varchar(100)", primary_key=False, default=None):
super(StringField, self).__init__(name, col_type, primary_key, default)
class IntegerField(Field):
def __init__(self, name=None, col_type="int(20)", primary_key=False, default=None):
super(IntegerField, self).__init__(name, col_type, primary_key, default)
class TextField(Field):
def __init__(self, name=None, col_type="text", primary_key=False, default=None):
super(TextField, self).__init__(name, col_type, primary_key, default)
创建数据表
创建数据表与flask_sqlalchemy类似采用模型类的方式，创建操作在除了create_all创建所有表以外，我还添加了一个create方法，对单个表进行创建（默认字符集为utf8）
创建成功后会打印一条消息
@classmethoddef create(cls):
info = {}
for k, v in cls.__mapping__.items():
info[k] = [v.primary_key, v.col_type, v.default]
sql = "create table " + cls.__tablename__ + "("
primary = None
for k, v in info.items():
col_info = "%s %s not null" % (k, v[1])
if v[0]:
primary = k
if v[2] is not None:
col_info += " default '%s'" % v[2] if type(v[2]) == str else " default %d" % v[2]
sql += col_info + ","
sql += "primary key(`%s`))engine=innodb default charset=utf8;" % primary
rows, result = cls.db.execute_sql(sql)
if not rows:
print("Create table `%s` (in database `%s`) success!" % (cls.__tablename__, cls.__database__))
插入
与flask_sqlalchemy类似，先创建一个模型类对象，再对该对象调用insert方法，该函数返回受影响的行数
def insert(self):"""
将调用该方法的对象作为一条记录加入相应的表中
如
obj = User(id=1,name='yzx')
obj.insert()
:return: 插入是否成功，成功则返回1
"""
# 组成实参列表
insert_values = [self.getValue(key) for key in self.__fields__]
insert_values.insert(0, self.getValue(self.__primary_key__))
# 将问号占位符替换成%s
sql = self.__insert__.replace("?", "%s")
# 执行插入语句
rows, results = self.db.execute_sql(sql, insert_values)
return rows
更新
更新为类方法，对模型类直接调用
@classmethoddef setValue(cls, id, col, value):
"""
更新数据库的值
调用
ClassName.setValue(id,col,value)
:param id: id值
:param col: 要更新的列名
:param value: 要更新的目标值
:return: 是否更新成功，成功返回1否则0
"""
if type(value) == int:
sql = "update %s set %s=%d where id=%d" % (cls.__tablename__, col, value, id)
else:
sql = "update %s set %s='%s' where id=%d" % (cls.__tablename__, col, value, id)
rows, result = cls.db.execute_sql(sql)
if not rows:
raise RuntimeError("Can't find data where id = %d" % id)
return rows
查询
查询提供了获取所有数据，根据id获取数据，根据Filter获取数据，值得一提的是，查询方法除了根据id获取数据（返回字典）以外返回的都是字典列表，其中每个字典代表一个记录，键为字段名，值为字段值。
以查询所有数据为例
@classmethoddef getAll(cls):
"""
获取某个表的所有数据
返回数据格式为一个列表
列表的每一项为字典，键为列名。
:return:list
"""
rows, result = cls.db.execute_sql(cls.__select__)
cols = [cls.__primary_key__] + cls.__fields__
return toDict(cols, result)
def toDict(cols, result):
result_list = []
for t in result:
d = {}
for k, v in zip(cols, t):
d[k] = v
result_list.append(d)
return result_list
以上是 用werkzeug实现一个简单的python web框架 的全部内容， 来源链接： utcz.com/z/386466.html
