前面的笔记里面谈到了装饰器和正则表达式，这里我要实现一个简单的web服务器。具体到功能上，首先是可以通过浏览器访问，还有一点就是有一个简单的模版系统。
1. 网络部分
网络部分使用的twisted，这个代码从twisted一本书中拷出来的。不要觉得很复杂照抄即可，我也不懂twisted。这里要说的是HTTP协议每行数据都是以\r\n结尾，所以这里协议继承自basic.LineReceiver。
from twisted.protocols import basic
from twisted.internet import protocol, reactor
from template import render_template
class HTTPProtocol(basic.LineReceiver):
def __init__(self,factory):
self.lines = []
self.factory = factory
def lineReceived(self, line):
self.lines.append(line)
if not line:
# self.sendResponse(self.factory.app.dispatch(lines))
self.sendResponse(self.factory.app.dispatch(self.lines))
def sendResponse(self,data):
self.sendLine(b"HTTP/1.1 200 OK")
self.sendLine(b"")
# responseBody = "You said:\r\n\r\n" + "\r\n".join(self.lines)
responseBody = data.encode('utf-8')+b'\r\n'
self.transport.write(responseBody)
self.transport.loseConnection()
class HTTPFactory(protocol.ServerFactory):
def __init__(self,app):
self.app = app
def buildProtocol(self, addr):
return HTTPProtocol(self)
2 路径装饰器
前面我谈到过装饰器的问题，每个路径对应的函数也可以使用装饰器来实现。把路径与带路径装饰器的函数关联起来，这里简单使用dict这种数据结构。实现的代码在下面列出，这里可以看到route函数是一个成员函数，最重要的一行代码就是self.routes[url] = func。本来装饰器只需要这行代码就能工作，但是如果装饰器不返回一个函数，这里说的是wrapper这个函数，那么就不能在一个函数上使用多个路径装饰器。
class Dongge():
def __init__(self):
self.routes = {}
def dispatch(self,lines):
line = lines[0].decode('utf-8')
method,url,version = line.split(' ')
print(line)
print(self.routes)
if url in self.routes:
return self.routes[url]()
return 'No such url resource'
def route(self,url):
def df(func):
self.routes[url] = func
def wrapper(*args, **kwargs):
return func(*args,**kwargs)
return wrapper
return df
app = Dongge()
@app.route('/')
@app.route('/index')
def index():
return 'index'
3 模板
模版在前面已经谈到过这方面的内容，这里只需要对render_template封装一下即可使用。
def view(self,tname,context):
path = '{}/{}.html'.format(self.templatedir,tname)
try:
with open(path,'r') as f:
html = f.read()
print(html)
return render_template(html,context)
except Exception as e:
print(e)
return ''
4 补充
让代码运行起来，其实就是简单的两句代码：
reactor.listenTCP(8000, HTTPFactory(app))
reactor.run()
如果你来跑这代码会存在什么问题，当然代码都运行过了肯定没多大问题。但是这里还有个让人很苦恼的事情，就是每次修改代码之后总得重新运行服务器。也许是说，我需要一个东西能监控我代码的改动，然后自动重启服务器。这个问题很其实很简单，看代码，这代码都是从网上拷的稍微修改了一下。
#start.py
# -*- coding:utf-8 -*-
from watchdog.observers import Observer
from watchdog.events import *
import time
from server import start
from multiprocessing import Process
class FileEventHandler(FileSystemEventHandler):
def __init__(self):
FileSystemEventHandler.__init__(self)
def on_moved(self, event):
if event.is_directory:
print("directory moved from {0} to {1}".format(event.src_path,event.dest_path))
else:
print("file moved from {0} to {1}".format(event.src_path,event.dest_path))
def on_created(self, event):
if event.is_directory:
print("directory created:{0}".format(event.src_path))
else:
print("file created:{0}".format(event.src_path))
def on_deleted(self, event):
if event.is_directory:
print("directory deleted:{0}".format(event.src_path))
else:
print("file deleted:{0}".format(event.src_path))
def on_modified(self, event):
if event.is_directory:
print("directory modified:{0}".format(event.src_path))
else:
print("file modified:{0}".format(event.src_path))
#如果改动的文件不是start.py则不需要重启服务器
if event.src_path.find('start.py') == -1:
self.startserver()
'''
启动服务器需要开启新进程，重启和启动代码放在一起了
'''
def startserver(self):
if hasattr(self,'p'):
self.p.terminate()
del self.p
p = Process(target=start)
p.start()
self.p = p
if __name__ == "__main__":
observer = Observer()
event_handler = FileEventHandler()
event_handler.startserver()
observer.schedule(event_handler,"./",True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
网友评论