我基本上是在尝试使用烧瓶、flask-restx、marshmallow 和 sqlalchemy 创建一个 API 来检索用户舰队的信息以及创建/添加新船只。
一个用户可以有多个船只，一个船只可以有多个行程，并且船只和行程的组合具有特定的燃料消耗。
虽然发送 GET 请求似乎会返回所有指定的消息，但发送 POST 请求会返回上述错误：
- -错误 - -
相关代码：
- -模型 - - -
一个容器模型，包括新容器的所有可能字段和一些用于查询和保存到数据库的辅助函数。
from zeus import db
from typing import List
class VesselModel(db.Model):
__tablename__ = "vessels"
imo = db.Column(db.Integer, primary_key = True)
name = db.Column(db.String(50), nullable=False)
breadth_m = db.Column(db.Float(precision=2), nullable=False)
depth_m = db.Column(db.Float(precision=2), nullable=False)
draught_m = db.Column(db.Float(precision=2), nullable=False)
baseline_m = db.Column(db.Float(precision=2), nullable=False)
length_ovrl_m = db.Column(db.Float(precision=2), nullable=False)
length_subm_m = db.Column(db.Float(precision=2), nullable=False)
lpp_m = db.Column(db.Float(precision=2), nullable=False)
lwl_m = db.Column(db.Float(precision=2), nullable=False)
ap_m = db.Column(db.Float(precision=2), nullable=False)
fp_m = db.Column(db.Float(precision=2), nullable=False)
# user_id = db.Column(db.Integer, db.ForeignKey('users.user_id'), nullable=False)
# user = db.relationship('UserModel')
#trips = db.relationship("TripModel", lazy="dynamic", primaryjoin="VesselModel.imo == TripModel.vessel_imo")
# fuel_consumption_tn_c02 = db.relationship("FuelConsumptionModel", lazy="dynamic", primaryjoin="VesselModel.imo == FuelConsumptionModel.vessel_imo")
def __init__(self, imo, name, breadth_m, depth_m, draught_m, baseline_m, length_ovrl_m, length_subm_m, lpp_m, lwl_m, ap_m, fp_m, user_id):
self.imo = imo
self.name = name
self.breadth_m = breadth_m
self.depth_m = depth_m
self.draught_m = draught_m
self.baseline_m = baseline_m
self.length_ovrl_m = length_ovrl_m
self.length_subm_m = length_subm_m
self.lpp_m = lpp_m
self.lwl_m = lwl_m
self.ap_m = ap_m
self.fp_m = fp_m
# self.user_id = user_id
def __repr__(self):
return 'VesselModel(imo=%s, name=%s, breadth_m=%s, depth_m=%s, draught_m=%s, baseline_m=%s, length_ovrl_m=%s, length_subm_m=%s, lpp_m=%s, lwl_m=%s, ap_m=%s, fp_m=%s' % (self.imo, self.name, self.breadth_m, self.depth_m, self.draught_m, self.baseline_m, self.length_ovrl_m, self.length_subm_m, self.lpp_m, self.lwl_m, self.ap_m, self.fp_m)
def json(self):
return {'imo': self.imo, 'name': self.name, 'breadth_m': self.breadth_m, 'depth_m': self.depth_m, 'draught_m': self.draught_m, 'baseline_m': self.baseline_m, 'length_ovrl_m': self.length_ovrl_m, 'length_subm_m': self.length_subm_m, 'lpp_m': self.lpp_m, 'lwl_m': self.lwl_m, 'ap_m': self.ap_m, 'fp_m': self.fp_m}
@classmethod
def find_by_imo(cls, imo) -> "VesselModel":
return cls.query.filter_by(imo=imo).first()
@classmethod
def find_all(cls) -> "VesselModel":
return cls.query.all()
# @classmethod
# def find_fleet(cls, user_id) -> List["VesselModel"]:
# return cls.query.filter_by(user_id=user_id).all()
def save_to_db(self) -> None:
db.session.add(self)
db.session.commit()
def delete_from_db(self) -> None:
db.session.delete(self)
db.session.commit()
---架构---
from ma import ma
from models.User import UserModel
from models.Vessel import VesselModel
from schemas.Trip import TripSchema
class VesselSchema(ma.SQLAlchemyAutoSchema):
trips = ma.Nested(TripSchema, many=True)
class Meta:
model: VesselModel
load_instance = True
include_fk = True
---资源---
class Fleet(Resource):
@fleet_ns.doc('Get all the vessels in the fleet')
def get(self):
return fleet_schema.dumps(VesselModel.find_all()), 200
@fleet_ns.expect(vessel)
@fleet_ns.doc('Add a vessel')
def post(self):
vessel_json = request.get_json()
if not vessel_json:
return {"message": "No input data provided"}, 400
try:
vessel_data = vessel_schema.load(vessel_json)
except ValidationError as err:
print("Im here")
return err.messages, 422
vessel_data.save_to_db()
return vessel_schema.dump(vessel_data), 201
---app.py----
from pydoc import doc
from flask import Flask, Blueprint, jsonify
from flask_restx import Api
from ma import ma
from zeus import db
from resources.vessel import Vessel, Fleet, vessel_ns, fleet_ns
from resources.trip import Trip, TripList, trip_ns, trips_ns
from resources.fuelconsumption import Fuel_Consumption, fuelconsumption_ns
from marshmallow import ValidationError
app = Flask(__name__)
bluePrint = Blueprint('api', __name__, url_prefix='/api')
api = Api(bluePrint, doc='/doc', title='Promilist - Hermes API')
app.register_blueprint(bluePrint)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///olympus.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['PROPAGATE_EXCEPTIONS'] = True
api.add_namespace(vessel_ns)
api.add_namespace(fleet_ns)
api.add_namespace(trip_ns)
api.add_namespace(trips_ns)
api.add_namespace(fuelconsumption_ns)
@app.before_first_request
def create_tables():
db.create_all()
@api.errorhandler(ValidationError)
def handle_validation_error(error):
return jsonify(error.messages), 400
vessel_ns.add_resource(Vessel, '/<int:vessel_imo>', endpoint =
'vessel_characteristics_by_imo')
fleet_ns.add_resource(Fleet, "", endpoint = 'fleet_by_user')
trip_ns.add_resource(Trip, '/<int:trip_id>', endpoint = 'trip_by_id')
trips_ns.add_resource(TripList, '', endpoint = 'trips_by_vessel')
fuelconsumption_ns.add_resource(Fuel_Consumption, '/<int:vessel_imo>/<int:trip_id>',
endpoint = 'fuel_consumption_by_vessel_and_trip_combination')
db.init_app(app)
if __name__ == '__main__':
ma.init_app(app)
app.run(port=5000, debug=True)
----文件树----
