Python使用Flask-RESTful开发接口时循环导入错误如何解决

问题根因 #

你遇到的是典型的Python循环导入问题，导入链路为：app.py导入security模块 → security导入User类 → User导入app中的db，但app.py中导入security的时候，db还没有完成初始化定义，所以导入失败。

修复步骤 #

1. 调整app.py的代码顺序，解决循环导入 #

将依赖User/security的导入逻辑，挪到db初始化完成之后，修改后的app.py完整代码如下：

from flask import Flask
from flask_mongoengine import MongoEngine
from flask_restful import Api
from flask_jwt import JWT
import urllib

app = Flask(__name__)

database_name = "API"
mongodb_password = ""
DB_URI = "mongodb+srv://m001-student:"+ urllib.parse.quote(mongodb_password) +"@cluster0.hr17o.mongodb.net/{}?retryWrites=true&w=majority".format(database_name)
app.config['MONGODB_HOST'] = DB_URI

app.secret_key = '#0#'
api = Api(app)

# 先完成db的初始化
db = MongoEngine()
db.init_app(app)

# db初始化完成后再导入security相关内容，避免循环
from security import auth,identity
jwt = JWT(app, auth, identity)#/auth

2. 修正security.py的方法调用错误 #

你在security中定义的findByName、findById是全局方法，不属于User类，不要用User.的方式调用，同时补充密码类型转换避免比对失败，修改后的代码：

from User import User
from werkzeug.security import safe_str_cmp

def auth(tel,password):
    user = findByName(tel)
    if user and safe_str_cmp(str(user.password), str(password)):
        return user

def identity(payload):
    id = payload['identity']
    return findById(id)

def findByName(number):
    customer_obj = User.objects(tel=number).first()
    if customer_obj:
        return customer_obj

def findById(customer_id):
    customer_obj = User.objects(customer_id=customer_id).first()
    if customer_obj:
        return customer_obj

3. 修正其他可运行问题 #

• 调整CustomerRepository.py的返回格式，避免返回响应对象无法做数据处理：

from User import User

class CustomerRepository():
    @staticmethod
    def findAll():
        return [user.to_json() for user in User.objects]

• 修正index.py的路由参数不匹配、运行入口判断问题：

from flask import request
from flask_restful import Resource
from flask_jwt import jwt_required
from app import app, api
from CustomerRepository import CustomerRepository as Repository

items = Repository.findAll()

class UserController(Resource):
    @jwt_required()
    def get(self, id):
        item = next(filter(lambda x: x['customer_id'] == id, items), None)
        return {'item': item}, 200 if item else 404
    
    def post(self):
        data = request.get_json()
        new_item = {'tel': data['tel'],'password':data['password']}
        items.append(new_item)
        return new_item
    
    def delete(self, id):
        global items
        items = list(filter(lambda x:x['customer_id']!=id,items))
        return items

    def put(self, id):
        data = request.get_json()
        item = next(filter(lambda x:x['customer_id'] == id, items),None)
        if item is None:
            item = {'tel':data['tel'], 'password':data['password']}
            items.append(item)
        else:
            item.update(data)

class ItemList(Resource):
    def get(self):
        return{'item':items}

# 修正路由参数和方法参数匹配
api.add_resource(UserController,'/items/<string:id>')
api.add_resource(ItemList,'/items')

if __name__ == '__main__':
    app.run(port=4000,debug=True)

内容的提问来源于stack exchange，提问作者Amoungui Serge