基於python的webapi專案結構的初探
0.上回學習了,flask官網推薦的”大型應用“,然後又結合前些年比較流行的MVC架構,我在探索一條合適自己習慣的專案結構。
1.先前的專案,是以實現功能為目標,這是所有軟體專案的核心利益。隨著時間推移,我逐漸把展示端,業務邏輯端,介面端,資料抽象實體端,都已從邏輯角度分離出來。這次我將繼續解耦,把獨立的層級從物理檔案角度切割。
2.為了實現這個目標,需要的這些知識支撐:python的模組,包,檔案,目錄,以及flask的藍圖 需要認真理解這幾個概念,在python中,目錄和包的概念極為相似,另外import 和 from...import 的使用方法也是極為靈活。
3.本次改造,是基於”FLASK登入的大型應用初探“的檔案構造基礎上。按照邏輯層,資料層,路由層。重新歸類程式碼邏輯。
這個目錄是參考mvc思路,其中contoller目錄存放所有業務邏輯,比如登入相關驗證都在此資料夾內(其實在python中這叫做包)。models資料夾存放所有資料實體,所有orm都需要資料實體。routes存放的是向前端提供的路由表。utils是公共函式庫檔案。
4.contoller,models,routes之所以解耦出這三個包(資料夾),目的在於收斂問題。這要說到一個軟體專案各個環節互動的頻次。對於後端程式設計師來說,最容易穩定的是和前端互動的介面(這當然是理想狀態,現實狀況是,後端程式設計師和前端程式設計師撕逼的凶殘程度,就差扣對方enter鍵了)。對這個框架來說就是routes層。在存在合適產品經理以及架構師的前提下,最先穩定下來的就是 前後端介面。由於前後端互動介面穩定下來後,業務也就逐步收斂穩定,這是代表業務功能的核心--資料庫結構 也水到渠成。那麼models層自然就創建出來。因此,後端程式設計師長期工作的區域就是contoller層。
5.這麼一分析,這種結構簡直完美。那麼就來試試,新增一個新功能需要幹什麼呢?上次內容只有登入的相關邏輯,這次我把資料庫的建立和資料插入也加入程式碼了。首先思考業務需求:用程式碼方式實現資料表建立,以及初始化資料。
models層已經有了資料實體。只需在controller層編寫createDB的程式碼就行。初始化資料當然也要在controller層撰寫。而後,需要提供人機互動介面,此任務自然能由router層完成。下面把所有程式碼貼出來。
5.1controller層 的 __init__.py 檔案為空,但必須建立
import sys import time import jsonlogin.pyfrom flask import request,current_app,jsonify from flask_login import LoginManager,login_user,logout_user,login_required,current_user from .user import UserLogin login_manager = LoginManager() login_manager.login_view = None login_manager.init_app(current_app) @login_manager.request_loader def load_user_from_request(request): token = request.headers.get('Authorization') if token == None: return None payload = UserLogin.verfiyUserToken(token) if payload != None: alternativeID = payload['data']['alternativeID'] sessionID = payload['data']['sessionID'] user = UserLogin.queryUser(alternativeID=alternativeID,sessionID=sessionID) else: user = None return user def hello(): returnData = {'code': 0, 'msg': 'success', 'data': 'hello login...' } return jsonify(returnData),200 @login_required def firstPage(): returnData = {'code': 0, 'msg': 'success', 'data': {'token':current_user.token,'tips':'First Blood(來自' + current_user.userName +')'}} return returnData,200 def login(): if request.method == 'POST': username = request.form['username'] password = request.form['password'] user = UserLogin.queryUser(userName = username) if (user != None) and (user.verifyPassword(password)) : login_user(user) returnData = {'code': 0, 'msg': 'success', 'data': {'token':user.token}} return jsonify(returnData),200 else : returnData = {'code': 1, 'msg': 'failed', 'data': {'tips':'username or password is not correct'} } return jsonify(returnData),200 @login_required def logout(): userName = current_user.userName # alternativeID = current_user.alternativeID sessionID = current_user.sessionID UserLogin.dropSessionID(sessionID) logout_user() returnData = {'code': 0, 'msg': 'success', 'data': {'tips':'Bye ' + userName} } return json.dumps(returnData),200 @login_required def changePws(): user = UserLogin.queryUser(userID = current_user.id) user.changePws() returnData = {'code': 0, 'msg': 'success', 'data': {'tips':'password was changed'} } return json.dumps(returnData),200
from flask import jsonify from models.database import init_db,db_session from models.operator import Operator def initdb(): init_db() returnData = {'code': 0, 'msg': 'success', 'data': 'init_db' } return jsonify(returnData),200 def addOper(): oper1 = Operator(1,'admin', '123') oper2 = Operator(2,'guest', '123') db_session.add(oper1) db_session.add(oper2) db_session.commit() return 'addOper' def verifyPassword(pws): oper = Operator.query.all() return str(oper.verifyPassword(pws))testdb.py
import uuid from flask_login import UserMixin from werkzeug.security import check_password_hash,generate_password_hash from datetime import datetime,timedelta,time from utils.jwttoken import genToken,verfiyToken from models.database import db_session from models.operator import Operator from models.operatorsession import OperatorSession class UserLogin(UserMixin):#之所以命名為UserLogin,只是為了區分資料庫的User表。此類僅僅是為了登陸管理而存在 def __init__(self,operater,sessionID=None): self.id = operater.id self.userName = operater.username self.alternativeID = operater.alternativeID self.oper = operater self.sessionID = None self.token = None exp = datetime.utcnow() + timedelta(seconds=60) self.genSessionID(exp,sessionID) self.genToken(exp) def get_id(self): return self.id def get(user_id): if not user_id: return None user = Operator.query.filter_by(id= userID).first() return UserLogin(user) def verifyPassword(self,password=None): return self.oper.verifyPassword(password) def genSessionID(self,exp,sessionID=None): if sessionID == None: self.sessionID = str(uuid.uuid4()) os = OperatorSession(sessionID= self.sessionID,exp_utc= exp,operator= self.oper) db_session.add(os) # self.oper.sessions.append(OperatorSession(sessionID= self.sessionID,exp_utc= exp)) db_session.commit() else: self.sessionID = sessionID def genToken(self,exp): token = genToken(exp,{'alternativeID':self.alternativeID,'sessionID':self.sessionID}) self.token = token return token @staticmethod def queryUser(**kwargs): if 'userName' in kwargs: username = kwargs['userName'] user = Operator.query.filter_by(username = username).first() return UserLogin(user) elif ('alternativeID' in kwargs) and ('sessionID' in kwargs): alternativeID = kwargs['alternativeID'] sessionID = kwargs['sessionID'] user = db_session.query(Operator).filter_by(alternativeID=alternativeID).join(OperatorSession).filter_by(sessionID=sessionID).first() if user: return UserLogin(user,sessionID) else: return None @staticmethod def verfiyUserToken(token): payload = verfiyToken(token) if not payload : removeSessions = db_session.query(OperatorSession).filter(OperatorSession.exp_utc < datetime.utcnow()).delete(synchronize_session=False) db_session.commit() return payload @staticmethod def dropSessionID(sessionID): removeSession = db_session.query(OperatorSession).filter_by(sessionID=sessionID).first() db_session.delete(removeSession) db_session.commit()user.py
5.2 models層 的 __init__.py 檔案為空,但必須建立
from sqlalchemy import create_engine from sqlalchemy.orm import scoped_session, sessionmaker from sqlalchemy.ext.declarative import declarative_base engine = create_engine("mysql+mysqlconnector://root:123@localhost:3306/hello_login", encoding="utf-8",echo=True) db_session = scoped_session(sessionmaker(autocommit=False,autoflush=False,bind=engine)) Base = declarative_base() Base.query = db_session.query_property() def init_db(): # 在這裡匯入定義模型所需要的所有模組,這樣它們就會正確的註冊在 # 元資料上。否則你就必須在呼叫 init_db() 之前匯入它們。 import models Base.metadata.create_all(bind=engine)database.py
from sqlalchemy import ForeignKey,Column, Integer, String, Date,BigInteger,DateTime from sqlalchemy.orm import relationship,backref from werkzeug.security import check_password_hash,generate_password_hash from .database import Base import uuid class Operator(Base): __tablename__ = 'operators' id = Column(Integer, primary_key=True) username = Column(String(50), unique=True,nullable=False) password = Column(String(200), unique=False,nullable=False) alternativeID = Column(String(200), unique=True,nullable=False) def __init__(self, id=None,username=None, password=None): self.id = id self.username = username self.password = generate_password_hash(username + password) #加鹽加密函式,通過隨機產生不同salt(鹽粒)混入原文,使每次產生的密文不一樣。 self.alternativeID = str(uuid.uuid1()) def __repr__(self): return '<operator %r>' % (self.username) def verifyPassword(self,password=None): if password is None: return False return check_password_hash(self.password,self.username + password)operator.py
from sqlalchemy import ForeignKey,Column, Integer, String, Date,BigInteger,DateTime from sqlalchemy.orm import relationship,backref from .database import Base from .operator import Operator class OperatorSession(Base): __tablename__ = 'oper_sessions' id = Column(Integer, primary_key=True) sessionID = Column(String(200), nullable=False) exp_utc = Column(DateTime, nullable=False) operator_id = Column(Integer, ForeignKey('operators.id'),nullable=False) operator = relationship(Operator,backref=backref('operators',uselist=True,cascade='delete,all'))operatorsession.py
5.3 router層 的 __init__.py 已經有關鍵程式碼了。
def init_app(app): with app.app_context(): from .home import bp,helloworld app.register_blueprint(bp) app.add_url_rule("/", view_func=helloworld) from .login import bp app.register_blueprint(bp) from .testdb import bp app.register_blueprint(bp)__init__.py
from flask import Blueprint,jsonify bp = Blueprint('home_page',__name__) @bp.route('/hello') def helloworld(): returnData = {'code': 0, 'msg': 'success', 'data': 'hello world' } return jsonify(returnData),200home.py
from flask import Blueprint import controller.login as lgi bp = Blueprint('login_page',__name__) @bp.route('/hello2') def hello(): return lgi.hello() @bp.route('/firstPage') def firstPage(): return lgi.firstPage() @bp.route('/login', methods=['POST']) def login(): return lgi.login() # @bp.route('/logout') # def logout(): # return login.logout()login.py
from flask import Blueprint,jsonify import controller.testdb as tb bp = Blueprint('testdb_page',__name__) @bp.route('/initdb') def initdb(): return tb.initdb() @bp.route('/addoper') def addOper(): return tb.addOper()testdb.py
5.3 工具utils層 的__init__.py 檔案為空,但必須建立
import jwt from jwt import PyJWTError from datetime import datetime,timedelta SECRECT_KEY = b'\x92R!\x8e\xc6\x9c\xb3\x89#\xa6\x0c\xcb\xf6\xcb\xd7\xbc' def genToken(exp,data): payload = { 'exp': exp, 'data': data } token = jwt.encode(payload,key= SECRECT_KEY,algorithm= 'HS256') return bytes.decode(token) def verfiyToken(tokenStr): try: tokenBytes = tokenStr.encode('utf-8') payload = jwt.decode(tokenBytes,key= SECRECT_KEY,algorithm= 'HS256') return payload except PyJWTError as e: print("jwt驗證失敗: %s" % e) return Nonejwttoken.py
5.4 專案主目錄的__init__.py 撰寫了一些引導語句
import sys,os sys.path.append(os.path.dirname(__file__)) from flask import Flask,jsonify from flask_cors import CORS def create_app(test_config=None): app = Flask(__name__) app.secret_key =b'\x15f\x07\xd3\xd9\xbf*\x82\xd1\xe6\xb4\xf2\x95\xdd\x8f\x12' #命令列中執行後拷貝出隨機值 python -c "import os; print(os.urandom(16))" CORS(app,supports_credentials=True) import hellologin.routes routes.init_app(app) return app__init__.py
5.5 setup.py 和上次一樣
from setuptools import setup setup( name='hellologin', version='1.0', long_description=__doc__, packages=['hellologin'], include_package_data=True, zip_safe=False, install_requires=['Flask'] )setup.py
6. 目前本專案已經有兩個功能,一是登入,二是初始化資料庫。可以看出來,從前幾次前端/後端/資料端的分離,到今天后端程式碼邏輯分離。我一直向著解耦這條路狂奔而去。
7. 這個框架結構,初步感受是,層級鮮明,合適瀑布開發模式,或者迭代思路明確的專案。但是呢,我作為獨立開發者,構思規劃並不那麼周全,在需要新增新功能時,就會在多個層來回跳轉,目前才十幾個py檔案,我不敢想當py檔案膨脹到幾十個上百後,我該如何定位。也許可以再每個層下,以業務功能定義目錄結構,這樣就能減輕檔案混亂壓力。究竟該怎麼做呢?
8.剛才給自己留下了一個疑問,目前我沒有很好的解決方案。這反倒讓我產生了另一個專案架構的思路,就是以產品功能作為解耦依據,暫且擱置。
9.很自然我現在需要一個使用者管理頁面。下一步,主要學習中心轉入使用者管理頁面的搭建。之後就可以涉及前端路由管理。這些都完成了,就要開始對上個問題探索答案。