写在前面
初次阅读此书是三年前,当时没经历过完整的项目 觉得这书就是扯淡 后来经历过项目加班与毒打 今天再翻开此书 觉得实乃不可多得之物 花些时间啃下来吧
需求
需求文档
写文档,列举需要实现的功能,详细列举,不考虑技术实现细节
需求评审与分析
主要是将需求文档落实到技术细节,评审需求需要的技术栈,然后评审需求是否可实现,预估每个子需求的工作量等
此处可以考虑后续的衍生需求,考虑技术实现是否可行是否困难,技术需要的工作量等
功能分析
技术人员对需求评审的结果进行技术实现分析,模块划分等
模块划分可以基于数据实体制作ER图,或者建立UML图
模块划分
作用是将一个大项目分成几个小模块,让手下的人去按模块开发
框架基础和技术选型
需要选择 语言 框架 数据库 然后考虑团队开发与实现能力等
wsgi
wsgi,全称Web Server Gateway Interface,Web服务器网关接口,是用来规定web server应如何和程序交互的网关协议。可以理解为一个web应用的容器,适配了程序和操作系统之间功能,将操作系统一些功能抽象为接口提供给程序使用
可以使用wsgi,目的是使用实现统一协议的web server,不然换着乱
简单的web server
一个基本的socket监听程序
# coding:utf-8import socketEOL1 = b'\n\n'
EOL2 = b'\n\r\n'
body = 'hello, world<h1> from tjh </h1>'
resp_params = ['HTTP/1.0 200 OK','Date: Sun, 31 jul 2024 09:35:33 GMT','Content-Type: text/html; charset=utf-8','Content-Length: ()\r\n'.format(len(body.encode())),body
]resp = '\r\n'.join(resp_params)def handle_connection(conn, addr):req = b''while EOL1 not in req and EOL2 not in req:req += conn.recv(1024)print(req)conn.send(resp.encode())conn.close()def main():ss = socket.socket(socket.AF_INET, socket.SOCK_STREAM)ss.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)ss.bind(('127.0.0.1', 8000))ss.listen(5) # conn max queue numberprint('http://127.0.0.1:8000')try:while True:conn, addr = ss.accept()handle_connection(conn, addr)finally:ss.close()if __name__ == '__main__':main()
效果
注意 Content-Type为text/plain 还是text/html
多线程版web server
还是阻塞模式,非阻塞会报错 还没处理
# coding:utf-8import socket
import errno
import threading
import timeEOL1 = b'\n\n'
EOL2 = b'\n\r\n'
body = 'hello, world<h1> from tjh </h1>-from {thread_name}'
resp_params = ['HTTP/1.0 200 OK','Date: Sun, 31 jul 2024 09:35:33 GMT','Content-Type: text/html; charset=utf-8','Content-Length: {length}\r\n',body
]resp = '\r\n'.join(resp_params)def handle_connection(conn, addr):req = b''while EOL1 not in req and EOL2 not in req:req += conn.recv(1024)print(req)current_thread = threading.currentThread()content_length = len(body.format(thread_name=current_thread.name).encode())print(current_thread.name)conn.send(resp.format(thread_name=current_thread.name, length=content_length).encode())conn.close()def main():ss = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# ss.setblocking(0) # set socket mode as non block # ss.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)ss.bind(('127.0.0.1', 8000))ss.listen(10) # conn max queue numberprint('http://127.0.0.1:8000')# ss.setblocking(0) # set socket mode as non block try:i = 0while True:try:conn, addr = ss.accept()except socket.error as e:if e.args[0] != errno.EAGAIN:raisecontinuei += 1print(i)t = threading.Thread(target=handle_connection, args=(conn, addr), name='thread-%s' % i)t.start()finally:ss.close()if __name__ == '__main__':main()
效果
简单wsgi application
wsgi协议分为两部分,一个是web server,一个是web application。接受请求时,会通过wsgi协议将数据发给web application,application处理完后,设置对应状体和header,之后返回body给web server,web server拿到数据后,进行http协议封装,返回完整http response
# coding: utf-8import os
import sysfrom app import simple_appdef wsgi_to_bytes(s):return s.encode()def run_with_cgi(app):environ = dict(os.environ.items())environ['wsgi.input'] = sys.stdin.bufferenviron['wsgi.errors'] = sys.stderrenviron['wsgi.version'] = (1, 0)environ['wsgi.multithread'] = Falseenviron['wsgi.multiprocess'] = Trueenviron['wsgi.run_once'] = Trueif environ.get('HTTPS', 'off') in ('on', '1'):environ['wsgi.url_scheme'] = 'https'else:environ['wsgi.url_scheme'] = 'http'headers_set = []headers_sent = []def write(data):out = sys.stdout.bufferif not headers_set:raise AssertionError('write() before start_response()')elif not headers_sent:status, resp_headers = headers_sent[:] = headers_setout.write(wsgi_to_bytes('Status: %s\r\n' % status))for header in resp_headers:out.write(wsgi_to_bytes('%s: %s\r\n' % header))out.write(wsgi_to_bytes('\r\n'))out.write(data)out.flush()def start_response(status, resp_headers, exc_info=None):if exc_info:try:if headers_sent:raise (exc_info[0], exc_info[1], exc_info[2])finally:exc_info = Noneelif headers_set:raise AssertionError('headers already set')headers_set[:] = [status, resp_headers]return writeresult = app(environ, start_response)try:for data in result:if data:write(data)if not headers_sent:write('')finally:if hasattr(result, 'close'):result.close()if __name__ == '__main__':run_with_cgi(simple_app)
理解wsgi
wsgi规定,application必须是一个可调用对象,则这个可调用对象可以是函数或实现了__call__方法的实例
wsgi中间件和werkzeug
flask
截至目前 有了两种方法提供web服务:直接通过socket处理请求,或者通过实现wsgi application部分协议
入门推荐
py微型框架有比如web.py, bottle,flask等 flask是一个不错的微型框架
tornado
高性能。tornado不是基于wsgi协议的框架,但提供了wsgi的支持,特性是异步和非阻塞。可以使用自带的http server进行部署而不是wsgi,因为wsgi是一个同步接口
和flask相比,tornado更侧重性能,整体并不比flask丰富,flask更多的支持对业务的满足
django
和微型框架不同,django框架不是仅需要两三个py文件就能跑起来的web框架,django功能更全也更大
django起步
管理系统后台开发
先安装django,再django-admin startproject project_name创建初始项目
cd到project下,创建一个app
models.py
from django.db import models# Create your models here.
class Student(models.Model):SEX_ITEMS = [(1, '男'), (2, '女'), (0, '未知')]STATUS_ITEMS = [(0, '申请'), (1, '通过'), (2, '拒绝')]name = models.CharField(max_length=128, verbose_name='姓名')sex = models.IntegerField(choices=SEX_ITEMS, verbose_name='性别')profession = models.CharField(max_length=128, verbose_name='职业')email = models.EmailField(verbose_name='Email')qq = models.CharField(max_length=128, verbose_name='QQ')phone = models.CharField(max_length=128, verbose_name='电话')status = models.IntegerField(choices=STATUS_ITEMS, default=0, verbose_name='审核状态')created_time = models.DateTimeField(auto_now_add=True, editable=False, verbose_name='创建时间')def __str__(self):return '<Student: {}>'.format(self.name)class Meta:verbose_name = verbose_name_plural = '学员信息'
admin.py
from django.contrib import admin# Register your models here.
from .models import Studentclass StudentAdmin(admin.ModelAdmin):list_display = ('id', 'name', 'sex', 'profession', 'email', 'qq', 'phone', 'status', 'created_time')list_filter = ('sex', 'status', 'created_time')search_fields = ('name', 'profession')fieldsets = ((None, {'fields': ('name',('sex', 'profession'),('email', 'qq', 'phone'),'status')}))admin.register(Student, StudentAdmin)
将创建的app在setting中加到installed_apps下
创建数据库迁移文件 python manage.py makemigrations
创建表 python manage.py migrate
创建超级用户 python manage.py createsuperuser
运行测试服务器,访问127.0.0.1:8000 发现是默认页面
访问127.0.0.1:8000/admin会跳转到刚开发的管理员页面,然后用刚创的管理员用户名和密码登录
管理员页面语言是英文,时区也是UTC,可在setting进行配置
重新启动测试服务器,再次登入管理员页面,发现语言变成中文