Task 09 — Build a Flask-RESTX Endpoint from Scratch

Difficulty: Intermediate
Goal: Build a complete, production-style Flask-RESTX endpoint that covers every major concept: Namespace, Resource, request parsers, POST body models, response models, marshalling, @jwt_required(), and Swagger documentation.

---

What you'll learn

• How to define a Namespace and wire it into the app
• How to write a reqparse parser for header and query arguments
• How to define a POST body model with @namespace.expect()
• How to define response models and use @namespace.response()
• How to marshal a response with @namespace.marshal_with()
• How everything looks in the Swagger UI

---

The feature you'll build

A Unit Summary endpoint that returns a quick summary for a given unit:

Method	Path	Auth
GET	/api/user/unitSummary	JWT required
POST	/api/user/unitSummary/flag	JWT required

• GET — accepts unit_id and start_date / end_date as headers, returns a typed summary response
• POST — accepts a JSON body to flag a unit for review, returns a success message

---

Step 1 — Define response models (app/models/unit_summary_models.py)

Response models go in app/models/ so they can be imported by the route.

from flask_restx import fields, Model

UnitSummaryModel = Model('UnitSummary', {
    'unit_id':        fields.String(description='Unit identifier'),
    'unit_name':      fields.String,
    'total_flow':     fields.Float(description='Total flow in m³'),
    'avg_flow':       fields.Float,
    'start_date':     fields.String,
    'end_date':       fields.String,
    'status':         fields.String,
})

FlagSuccessModel = Model('FlagSuccess', {
    'message': fields.String,
    'status':  fields.String,
    'unit_id': fields.String,
})

---

Step 2 — Create the namespace and parsers (app/routes/unit_summary.py)

from flask_restx import Namespace, Resource, reqparse, fields, marshal
from flask_jwt_extended import jwt_required, current_user
from app.models.unit_summary_models import UnitSummaryModel, FlagSuccessModel

unitSummaryNamespace = Namespace('Unit Summary', path='/unitSummary')

# --- Register models on the namespace ---
unitSummaryNamespace.add_model(UnitSummaryModel.name, UnitSummaryModel)
unitSummaryNamespace.add_model(FlagSuccessModel.name, FlagSuccessModel)

# --- GET parser: parameters come as headers ---
get_parser = reqparse.RequestParser()
get_parser.add_argument('unit_id',    type=str, required=True,  location='headers', help='Unit ID is required')
get_parser.add_argument('start_date', type=str, required=True,  location='headers', help='Start date DD/MM/YYYY')
get_parser.add_argument('end_date',   type=str, required=True,  location='headers', help='End date DD/MM/YYYY')

# --- POST body model: documented in Swagger as the request body ---
FlagPayload = unitSummaryNamespace.model('FlagPayload', {
    'unit_id': fields.String(required=True,  description='Unit to flag'),
    'reason':  fields.String(required=True,  description='Reason for flagging'),
    'severity': fields.String(required=False, description='low | medium | high', default='medium'),
})

---

Step 3 — Write the GET route

@unitSummaryNamespace.route('')
class UnitSummaryRoute(Resource):

    @jwt_required()
    @unitSummaryNamespace.expect(get_parser)                             # documents headers in Swagger
    @unitSummaryNamespace.response(200, 'Success', model=UnitSummaryModel)
    @unitSummaryNamespace.response(400, 'Missing required header')
    @unitSummaryNamespace.response(401, 'Unauthorized')
    def get(self):
        args = get_parser.parse_args()                   # returns 400 automatically if required arg missing
        unit_id     = args['unit_id']
        start_date  = args['start_date']
        end_date    = args['end_date']
        industry_id = current_user['industryId']

        # Call your service here
        # summary = UnitSummaryService().get_summary(unit_id, start_date, end_date, industry_id)

        return marshal({
            'unit_id':    unit_id,
            'unit_name':  'Pump Station 1',
            'total_flow': 1234.56,
            'avg_flow':   51.44,
            'start_date': start_date,
            'end_date':   end_date,
            'status':     'success',
        }, UnitSummaryModel), 200

---

Step 4 — Write the POST route

@unitSummaryNamespace.route('/flag')
class UnitFlagRoute(Resource):

    @jwt_required()
    @unitSummaryNamespace.expect(FlagPayload, validate=True)             # validates body; 400 if invalid
    @unitSummaryNamespace.response(200, 'Flagged successfully', model=FlagSuccessModel)
    @unitSummaryNamespace.response(400, 'Validation error')
    def post(self):
        from flask import request
        data = request.json

        unit_id  = data['unit_id']
        reason   = data['reason']
        severity = data.get('severity', 'medium')

        # Call your service here
        # FlagService().flag_unit(unit_id, reason, severity, current_user['userId'])

        return marshal({
            'message': f'Unit {unit_id} flagged successfully',
            'status':  'success',
            'unit_id': unit_id,
        }, FlagSuccessModel), 200

validate=True on @expect

When validate=True is set, Flask-RESTX will reject the request with a 400 if any required=True field is missing from the JSON body. This is the fastest way to guard a POST endpoint without writing manual checks.

---

Step 5 — Register the namespace in the API blueprint

Open app/apis/user.py and add:

from app.routes.unit_summary import unitSummaryNamespace

api.add_namespace(unitSummaryNamespace)

That's all — Flask-RESTX picks up the routes and adds them to the Swagger UI automatically.

---

Step 6 — Verify in Swagger UI

1. Run the server: python app.py
2. Open http://localhost:5001/api/user/docs
3. Find the Unit Summary section
4. Click GET /unitSummary — you should see unit_id, start_date, end_date listed as header parameters
5. Click POST /unitSummary/flag — you should see the FlagPayload schema as the request body
6. Authenticate via the Authorize button (use your JWT), then execute both endpoints

---

What to check

• GET without a required header returns 400 with a descriptive message
• POST with a missing required field returns 400
• GET without a JWT returns 401
• The Swagger UI shows the correct parameter locations (header vs body)
• The response schema matches UnitSummaryModel / FlagSuccessModel in Swagger
• marshal(data, model) serializes only the fields defined in the model (extra keys are stripped)

---

Concepts recap

Concept	Where used
Namespace	Groups related routes under a path prefix; shown as a section in Swagger
Resource	Class-based view — one method per HTTP verb (get, post, put, delete)
reqparse	Parses and validates request arguments from headers, query string, or form
@namespace.expect(model, validate=True)	Documents and optionally validates the POST body
@namespace.response(code, desc, model)	Documents a response code and its shape in Swagger
marshal(data, model)	Serializes the return dict through a model — always used explicitly in the return statement
fields.Nested / fields.List	Compose complex typed response shapes
fields.Raw	Untyped escape hatch for variable-shape data

---

Further Reading

• Flask-RESTX full docs
• Request parsing
• Response marshalling
• Swagger / API documentation
• Flask-JWT-Extended