1. 程式人生 > 實用技巧 >基於python的webapi專案結構的初探

基於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 json
from 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
login.py
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),200
home.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 None
jwttoken.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.很自然我現在需要一個使用者管理頁面。下一步,主要學習中心轉入使用者管理頁面的搭建。之後就可以涉及前端路由管理。這些都完成了,就要開始對上個問題探索答案。