目錄
前言
在Tornado框架中嵌入我們的數據庫處理模塊。
1、在Tornado框架中嵌入我們的數據庫處理模塊
2、使用Postman來測試接口
結語
手冊目錄：
軟件下載及環境安裝：
實習手冊一（基於Tornado框架的接口響應服務）軟件下載與環境配置
Tornado基本框架搭建：
實習手冊二（基於Tornado框架的接口響應服務）Tornado基本框架的搭建
Tornado框架中日志的記錄，路由的分發，接口的響應：
實習手冊三（基於Tornado框架的接口響應服務）Tornado框架中日志的記錄，路由的分發，接口的響應
通過PyMySQL，SQLAlchemy在PyCharm中實現對數據的增刪改查：
實習手冊四（基於Tornado框架的接口響應服務）通過PyMySQL，SQLAlchemy在PyCharm中實現對數據的增刪改查
SqlAlchemy的再封裝，PyMySQL和SqlAlchemy的結合使用：
實習手冊五（基於Tornado框架的接口響應服務）SqlAlchemy的再封裝，PyMySQL和SqlAlchemy的結合使用
本章我們會將前幾章的功能結合起來，在Tornado框架中嵌入我們的數據庫處理模塊。
在前兩章中，我們已經實現了通過pymysql+sqlalchemy實現了使用python代碼來處理數據庫，而且已經對sqlalchemy進行了封裝，現在我們直接將上一章中的SqlAlBase，MysqlBase以及User類寫入我們的Tornado框架中：
import datetime
import json
from tornado.web import RequestHandler
from tornado.web import Application
from tornado.ioloop import IOLoop
import pymysql
from sqlalchemy import create_engine, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import MetaData
from sqlalchemy import Column, String, Integer
# URL用來作為SqlAlchemy創造引擎的url
# 格式為："mysql+pymysql://root:你的mysql密碼@localhost:3306/數據庫名字"
URL = "mysql+pymysql://root:你的密碼@localhost:3306/數據庫名稱"
class SqlAchBase():
engine = create_engine(URL)
base = declarative_base()
session = sessionmaker(bind=engine)()
metadata = MetaData()
@classmethod
def get_session(cls):
return cls.session
class MysqlBase():
session = SqlAchBase.get_session()
conn = None
# 測試連接方法
@classmethod
def TestConn(cls):
if not cls.conn:
cls.conn = pymysql.connect(user='root', host='localhost', password=你的密碼, db=數據庫名稱, port=3306)
print("trying to connect to '%s'" % cls.conn)
else:
print("already connected to '%s'" % cls.conn)
# 增
@classmethod
def add(cls, user):
cls.TestConn()
cls.session.add(user)
cls.session.commit()
# 刪
@classmethod
def delete(cls, filter):
cls.TestConn()
cls.session.query(cls).filter(text(filter)).delete(synchronize_session=False)
cls.session.commit()
# 改
@classmethod
def update(cls, filter, value):
cls.TestConn()
cls.session.query(cls).filter(text(filter)).update(value, synchronize_session=False)
cls.session.commit()
# 查
@classmethod
def query(cls, filter=None):
cls.TestConn()
if filter:
item_list = cls.session.query(cls).filter(text(filter)).all()
for item in item_list:
print(item)
else:
item_list = cls.session.query(cls)
for item in item_list:
print(item)
class User(SqlAchBase().base, MysqlBase):
__tablename__ = 'user'
userid = Column(Integer, autoincrement=True, primary_key=True)
uname = Column(String(30))
pwd = Column(String(30))
create_time = Column(String(30))
# 注冊功能處理器
class RegisterHandler(RequestHandler):
# 用來存儲接口的請求參數
data = None
def post(self, *args, **kwargs):
path = self.request.path.split('/')
# 注意，這裡是通過body傳遞了json格式的數據，tornado獲取後會將其自動轉化為字典
self.data = json.loads(self.request.body)
method = path[-1]
if callable(getattr(self, method)):
getattr(self, method)()
else:
self.write("404 not found")
def register_user(self):
user = User(uname=self.data['uname'], pwd=self.data['pwd'], create_time=str(datetime.datetime.now()))
User.add(user)
app = Application([
# 注冊功能路由
(r'/user/register/.*', RegisterHandler)
])
if __name__ == '__main__':
app.listen(8888)
IOLoop.current().start()可以看到，我們這裡寫了一個RegisterHandler（注冊處理器），用來響應注冊功能，其中有一個register_user()方法，用來作為接口，來實現用戶注冊的功能。
首先，我們要實現的是一個注冊用戶的功能，在前面我們已經提到過，Tornado框架中是通過路由來確定大的功能分區，再通過get或post方法來進行小功能的分發，其中，Tornado中實現小功能的部分是通過python代碼中的def方法來實現的，其實可以理解成，一個方法對應的就是一個功能，同時一個接口對應的就是一個方法，Tornado中是通過接口來調用對應的功能的。
因此，我們如果想對功能進行測試的話，首先要學會測試接口，而Postman便是一個非常簡易方便的用來測試接口的軟件，之前沒有了解過Postman的小伙伴們可以點擊下方這個鏈接進行學習。
接口測試之Postman使用全指南(原來使用 Postman測試API接口如此簡單)
首先，先進入我們的Workspaces界面，點擊左側欄加號來新建一個連接目錄
然後點擊右側的加號，來新建連接
然後將請求方式切換為Post，輸入url，在body部分選中row，然後將Text切換為json
然後輸入新注冊的用戶名，密碼即可，時間會直接獲取當前時間，詳細見上方的代碼
然後點擊send，即可發送請求，請求成功後結果欄會返回1，然後進入到navicat中進行查看
可以看到，已經成功添加了一條新數據，create_time格式可以通過後段的代碼進行調整，這裡不多做介紹。
本章我們將Tornado框架和Mysql服務結合起來，通過請求接口傳遞對應參數便可以對數據庫進行對應的處理，其他功能諸如刪、改、查，都可以依照這個樣式來實現，本章只是為了實現此功能而將代碼全部放在一個python文件中，略顯冗長請大家多多包容。
