pip install tornado

# 數(shù)據(jù)庫工具類
import peewee
import peewee_async
from datetime import datetime
# 創(chuàng)建數(shù)據(jù)庫鏈接對象
database = peewee_async.PooledMySQLDatabase("scoial", host="127.0.0.1",
                                            port=3306, user="root", password="123456")
# 基類
class BaseModel(peewee.Model):
    id = peewee.IntegerField(primary_key=True, unique=True, constraints=
    [peewee.SQL('AUTO_INCREMENT')])
    # 入庫時間
    create_time = peewee.DateTimeField(default=datetime.now,
                                       help_text="入庫時間")
    # 重寫父類方法
    def save(self, *args, **kwargs):
        # 判斷什么時候賦值入庫方法
        if self._pk is None:
            # 賦值
            self.create_time = datetime.now()
        return super(BaseModel, self).save(*args, **kwargs)
    class Meta:
        # 傳遞數(shù)據(jù)庫連接
        database = database
# 用戶表
class UserModel(BaseModel):
    email = peewee.CharField(null=False, unique=True, max_length=100)
    password = peewee.CharField(null=False, max_length=567)
    state = peewee.IntegerField(null=False, default=0)
    class Meta:
        # 聲明表明
        db_table = "user"
if __name__ == '__main__':
    # 創(chuàng)建表
    UserModel.create_table(True)
    # 刪除表
    # UserModel.drop_table(True)
    # 添加測試數(shù)據(jù)

import tornado.ioloop
import tornado.web
import redis
# 導(dǎo)入基礎(chǔ)類
from base import BaseHandler
# 導(dǎo)入數(shù)據(jù)庫操作類
from db import database
import os
from tornado.options import define, options
define('port', default=8000, help='default port',type=int)
import peewee_async
# import aioredis
import asyncio
# 導(dǎo)入用戶模塊
import user
# 設(shè)置tornado靜態(tài)文件夾目錄
static_path = os.path.join(os.path.dirname(__file__),"static")
class TestHandler(BaseHandler):
    async def get(self):
        self.finish({"msg":"hello Tornado"})
# 集成路由
urlpatterns = [(r'/',TestHandler)]
urlpatterns += (user.urlpatterns)
app = tornado.web.Application(
    handlers=urlpatterns,static_path=static_path,debug=True
)
# 將數(shù)據(jù)庫連接注入到事件循環(huán)中
app.objects = peewee_async.Manager(database)
app.redis = redis.Redis(password='123',decode_responses=True)
# 將異步redis鏈接注入到事件循環(huán)
# async def redis_pool(loop):
#
#     return await aioredis.create_redis_pool('localhost',minsize=1,
#                                             maxsize=1000,encoding='utf8',loop=loop)
#
# loop = asyncio.get_event_loop()
# app.redis = loop.run_until_complete(redis_pool(loop))
# app.objects = peewee_async.Manager(database)
if __name__ == "__main__":
    print("開啟服務(wù)")
    tornado.options.parse_command_line()
    app.listen(options.port)
    tornado.ioloop.IOLoop.instance().start()