欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

用Python徒手写一个web开发框架

程序员文章站 2022-06-13 17:02:27
...

自从转行做码农以来,零零碎碎总会参与web开发相关的工作,但一直都没系统地学习一下,现在处于离职前夕,刚好有时间,跟着廖雪峰大神《Python教程》的实战教程,从头开始写一个web开发框架,以理清其中的脉络。

Web App骨架

整个框架建立在asyncio的基础上,而异步IO的现实是用的协程模型,跟传统子程序(即函数,通过栈实现,一个线程就是执行一个子程序,最终一层一层返回给程序入口)相比,有两点优势:

a. 极高的执行效率,没有线程切换的开销,通过程序控制中断(与单片机里的中断类似)来切换任务

b. 不需要多线程的锁机制,因为只有一个线程

import logging; logging.basicConfig(level=logging.INFO)

import asyncio, os, json, time
from datetime import datetime

from aiohttp import web

def index(request):
    return web.Response(body=b'<h1>Awesome</h1>')

@asyncio.coroutine
def init(loop):
    app = web.Application(loop=loop)
    app.router.add_route('GET', '/', index)
    srv = yield from loop.create_server(app.make_handler(), '127.0.0.1', 9000)
    logging.info('server started at http://127.0.0.1:9000...')
    return srv

loop = asyncio.get_event_loop()
loop.run_until_complete(init(loop))
loop.run_forever()

ORM层

1. 创建一个全局的连接池

2. 实现Select和Insert, Update, Delete的SQL模板

3. 定义Field和各种Filed子类

Filed有四个属性,name,column_type, primary_key(type:boolean), default

4. 定义ModelMetaclass

因为操作不同的表要不同的对象,不同的对象需要不同的类来创建,而只有使用者才能根据表的结构定义出对应的类,即所有的类需要动态定义,所以这里选择用metaclass来创建类,即:

定义metaclass就可以创建类,再创建实例。可以把类看作是metaclass创建出来的实例。

需要实现__new__(cls, name, bases, attrs)方法,当传入metaclass时,它指示Python解释器在创建对象时,要通过ModelMetaclass.__new__()来创建,在此可以修改类的定义,比如,加上新的方法,然后,返回修改后的定义

class ModelMetaclass(type):

    def __new__(cls, name, bases, attrs):
        if name=='Model':
            return type.__new__(cls, name, bases, attrs)
        print('Found model: %s' % name)
        mappings = dict()
        for k, v in attrs.items():
            if isinstance(v, Field):
                print('Found mapping: %s ==> %s' % (k, v))
                mappings[k] = v
        for k in mappings.keys():
            attrs.pop(k)
        attrs['__mappings__'] = mappings # 保存属性和列的映射关系
        attrs['__table__'] = name # 假设表名和类名一致
        return type.__new__(cls, name, bases, attrs)

5. 定义基类Model

当用户定义一个class User(Model)时,Python解释器首先在当前类User的定义中查找metaclass,如果没有找到,就继续在父类Model中查找metaclass,找到了,就使用Model中定义的metaclass的ModelMetaclass来创建User类,也就是说,metaclass可以隐式地继承到子类。

class Model(dict, metaclass=ModelMetaclass):

    def __init__(self, **kw):
        super(Model, self).__init__(**kw)

    def __getattr__(self, key):
        try:
            return self[key]
        except KeyError:
            raise AttributeError(r"'Model' object has no attribute '%s'" % key)

    def __setattr__(self, key, value):
        self[key] = value

    def save(self):
        fields = []
        params = []
        args = []
        for k, v in self.__mappings__.items():
            fields.append(v.name)
            params.append('?')
            args.append(getattr(self, k, None))
        sql = 'insert into %s (%s) values (%s)' % (self.__table__, ','.join(fields), ','.join(params))
        print('SQL: %s' % sql)
        print('ARGS: %s' % str(args))

路由

1. 把一个函数映射为URL处理函数,eg.

def get(path):
    '''
    Define decorator @get('/path')
    '''
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kw):
            return func(*args, **kw)
        wrapper.__method__ = 'GET'
        wrapper.__route__ = path
        return wrapper
    return decorator

2, 编写一个add_route函数‘

def add_route(app, fn):
    method = getattr(fn, '__method__', None)
    path = getattr(fn, '__route__', None)
    if path is None or method is None:
        raise ValueError('@get or @post not defined in %s.' % str(fn))
    if not asyncio.iscoroutinefunction(fn) and not inspect.isgeneratorfunction(fn):
        fn = asyncio.coroutine(fn)
    logging.info('add route %s %s => %s(%s)' % (method, path, fn.__name__, ', '.join(inspect.signature(fn).parameters.keys())))
    app.router.add_route(method, path, RequestHandler(app, fn))

参数解析

1. 检查必填参数是否被传入

2. 将参数分装到dict结构中,方便在handlers中处理

middleware

即拦截器,可以改变URL的输入、输出,甚至可以决定不继续处理而直接返回。middleware的用处就在于把通用的功能从每个URL处理函数中拿出来,集中放在一个地方

1. 一个URL日志的logger可以简单定义如下:

@asyncio.coroutine
def logger_factory(app, handler):
    @asyncio.coroutine
    def logger(request):
        # 记录日志:
        logging.info('Request: %s %s' % (request.method, request.path))
        # 继续处理请求:
        return (yield from handler(request))
    return logger

2. 不同response的处理

@asyncio.coroutine
def response_factory(app, handler):
    @asyncio.coroutine
    def response(request):
        # 结果:
        r = yield from handler(request)
        if isinstance(r, web.StreamResponse):
            return r
        if isinstance(r, bytes):
            resp = web.Response(body=r)
            resp.content_type = 'application/octet-stream'
            return resp
        if isinstance(r, str):
            resp = web.Response(body=r.encode('utf-8'))
            resp.content_type = 'text/html;charset=utf-8'
            return resp
        if isinstance(r, dict):
            ...

配置文件

Python本身语法简单,可以直接用Python源代码来实现配置

如果要部署到服务器时,通常需要修改数据库的host等信息,直接修改config_default.py不是一个好办法,更好的方法是编写一个config_override.py,用来覆盖某些默认设置

configs = config_default.configs

try:
    import config_override
    configs = merge(configs, config_override.configs)
except ImportError:
    pass

异常处理

用一个整数表示错误码,这种方式很难维护错误码,客户端拿到错误码还需要查表得知错误信息。更好的方式是用字符串表示错误代码,不需要看文档也能猜到错误原因。

class APIError(Exception):
    def __init__(self, error, data="", message=""):
        self.error = error
        self.data = data
        self.message = message


class APIValueError(APIError):
    def __init__(self, field, message=""):
        super().__init__("value:invalid", field, message)

分页

class Page(object):
    """
    Page object for display pages
    """
    def __init__(self, item_count, page_index=1, page_size=10):
        self.item_count = item_count
        self.page_size = page_size
        self.page_count = item_count // page_size + (1 if item_count % page_size > 0 else 0)
        if (item_count == 0) or (page_index > self.page_count):
            self.offset = 0
            self.limit = 0
            self.page_index = 1
        else:
            self.page_index = page_index
            self.offset = (page_index - 1) * page_size
            self.limit = self.page_size
        self.has_next = self.page_index < self.page_count
        self.has_previous = self.page_index > 1

 

相关标签: web开发