Flask RESTful: ساخت API های پیشرفته

در دنیای پرشتاب توسعه وب، ارائه APIهای کارآمد و قابل اعتماد برای تعامل بین سرویس‌ها و برنامه‌های کاربردی مختلف، از اهمیت حیاتی برخوردار است. فریم‌ورک Flask به دلیل سبکی و انعطاف‌پذیری‌اش، انتخابی محبوب برای ساخت برنامه‌های وب کوچک تا متوسط و همچنین APIها محسوب می‌شود. با این حال، هنگامی که صحبت از ساخت APIهای RESTful پیشرفته و مقیاس‌پذیر به میان می‌آید، نیاز به ابزارهایی برای استانداردسازی فرآیندهایی مانند تجزیه درخواست‌ها، اعتبارسنجی داده‌ها و سریالیزیشن پاسخ‌ها احساس می‌شود. اینجاست که Flask-RESTful وارد عمل می‌شود.

Flask-RESTful یک اکستنشن برای Flask است که ابزارهایی قدرتمند برای ساخت سریع و آسان APIهای REST فراهم می‌کند. این کتابخانه بر پایه اصول REST بنا شده و با ارائه الگوهای توسعه‌ای واضح، به شما کمک می‌کند تا APIهایی با ساختار منطقی، قابل نگهداری و مقیاس‌پذیر بسازید. این ابزار با ساده‌سازی روال‌های معمول توسعه API، مانند مسیریابی منابع، مدیریت خطاهای HTTP و تجزیه آرگومان‌های درخواست، به توسعه‌دهندگان اجازه می‌دهد تا بر منطق تجاری اصلی تمرکز کنند. در این پست وبلاگ جامع، ما به تفصیل به بررسی Flask-RESTful خواهیم پرداخت و نحوه استفاده از آن برای ساخت APIهای پیشرفته و قدرتمند را از صفر تا صد، با مثال‌های عملی، آموزش خواهیم داد. هدف این راهنما، مسلح کردن شما به دانش و ابزارهایی است که برای توسعه APIهای RESTful با کیفیت بالا در محیط تولید نیاز دارید.

مقدمه: چرا Flask-RESTful؟

پیش از غواصی در جزئیات فنی، بیایید ابتدا به این سوال پاسخ دهیم که چرا باید از Flask-RESTful استفاده کنیم، در حالی که Flask به تنهایی نیز قابلیت ساخت API را دارد. Flask یک میکرو-فریم‌ورک است که به طور عمدی حداقل ابزارها را فراهم می‌کند و انتخاب‌ها را به توسعه‌دهنده و اکستنشن‌های جامعه واگذار می‌کند. این رویکرد برای انعطاف‌پذیری عالی است، اما برای ساخت APIهای پیچیده، ممکن است نیاز به نوشتن کدهای تکراری برای کارهایی مانند:

• تعریف endpointهای مختلف برای یک منبع (مثلاً GET برای لیست، GET برای آیتم خاص، POST برای ایجاد، PUT برای به‌روزرسانی، DELETE برای حذف).
• تجزیه و اعتبارسنجی داده‌های ورودی (مانند بررسی اینکه آیا یک فیلد الزامی است یا نوع داده آن صحیح است).
• سریالیزیشن داده‌های خروجی (تبدیل آبجکت‌های پایتون به فرمت JSON استاندارد).
• مدیریت خطاهای HTTP به صورت یکپارچه.

Flask-RESTful این فرآیندها را با ارائه کلاس‌ها و توابع مخصوص، به شکل قابل توجهی ساده می‌کند. این کتابخانه با معرفی مفهوم “Resource”، به شما اجازه می‌دهد تا تمام عملیات مربوط به یک منبع خاص را در یک کلاس واحد کپسوله کنید. این کار به سازماندهی بهتر کد کمک کرده و خوانایی و قابلیت نگهداری آن را افزایش می‌دهد. همچنین، با ابزارهایی مانند reqparse برای اعتبارسنجی درخواست‌ها و fields و marshal_with برای سریالیزیشن پاسخ‌ها، Flask-RESTful به شما اطمینان می‌دهد که API شما از استانداردها و الگوهای RESTful پیروی می‌کند.

مزایای اصلی استفاده از Flask-RESTful عبارتند از:

• سادگی و سرعت توسعه: با الگوهای توسعه‌ای مشخص، سرعت ساخت APIها افزایش می‌یابد.
• سازگاری با REST: به شما کمک می‌کند تا APIهای سازگار با اصول REST طراحی کنید.
• مدیریت آسان منابع: مفهوم Resource کد شما را سازمان‌دهی می‌کند.
• اعتبارسنجی قدرتمند: reqparse ابزاری قوی برای اعتبارسنجی پارامترهای درخواست فراهم می‌کند.
• سریالیزیشن یکپارچه: marshal_with و fields به استانداردسازی خروجی‌ها کمک می‌کنند.
• مدیریت خطاها: ارائه راهکاری یکپارچه برای مدیریت خطاهای HTTP.

به طور خلاصه، اگر به دنبال ساخت APIهای RESTful هستید که نه تنها سریع توسعه یابند بلکه ساختاری محکم، قابل نگهداری و مقیاس‌پذیر داشته باشند، Flask-RESTful ابزاری ضروری در جعبه ابزار شما خواهد بود.

اصول اولیه Flask-RESTful: ساختار و مؤلفه‌ها

Flask-RESTful بر پایه چند مفهوم کلیدی ساخته شده است که درک آن‌ها برای شروع کار ضروری است: API، Resource، و Request Parser. در ادامه به تفصیل به این مؤلفه‌ها و نحوه راه‌اندازی اولیه می‌پردازیم.

نصب و راه‌اندازی

قبل از هر چیز، نیاز داریم تا Flask و Flask-RESTful را نصب کنیم. توصیه می‌شود این کار را در یک محیط مجازی (virtual environment) انجام دهید.

pip install Flask Flask-RESTful

پس از نصب، می‌توانید یک فایل پایتون ایجاد کرده و شروع به کدنویسی کنید. ساختار پایه یک برنامه Flask-RESTful به شکل زیر است:

from flask import Flask
from flask_restful import Api, Resource

app = Flask(__name__)
api = Api(app)

# اینجا منابع (Resources) تعریف و به API اضافه می‌شوند
# api.add_resource(MyResource, '/my-resource')

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

در این قطعه کد:

• Flask(__name__): یک نمونه از برنامه Flask ایجاد می‌کند.
• Api(app): یک شیء Api از Flask-RESTful ایجاد می‌کند که برنامه Flask شما را می‌پذیرد. این شیء مسئول مدیریت مسیریابی و هندلینگ منابع RESTful است.
• Resource: کلاس پایه‌ای است که تمام منابع RESTful شما باید از آن ارث ببرند. هر متد HTTP (GET, POST, PUT, DELETE و غیره) که شما می‌خواهید پشتیبانی کنید، به عنوان یک متد در این کلاس تعریف می‌شود.

اولین API RESTful شما

بیایید یک مثال ساده برای درک نحوه کارکرد Resourceها بسازیم. فرض کنید می‌خواهیم یک API برای مدیریت لیستی از وظایف (tasks) بسازیم.

from flask import Flask
from flask_restful import Api, Resource

app = Flask(__name__)
api = Api(app)

tasks = {
    'task1': {'description': 'Write a blog post', 'done': False},
    'task2': {'description': 'Learn Flask-RESTful', 'done': True},
}

class TaskList(Resource):
    def get(self):
        return tasks

    def post(self):
        # اینجا منطق برای ایجاد یک تسک جدید اضافه می‌شود
        return {'message': 'Task created'}, 201

class Task(Resource):
    def get(self, task_id):
        if task_id not in tasks:
            return {'message': 'Task not found'}, 404
        return tasks[task_id]

    def put(self, task_id):
        # اینجا منطق برای به‌روزرسانی یک تسک اضافه می‌شود
        if task_id not in tasks:
            return {'message': 'Task not found'}, 404
        return {'message': f'Task {task_id} updated'}, 200

    def delete(self, task_id):
        if task_id not in tasks:
            return {'message': 'Task not found'}, 404
        del tasks[task_id]
        return {'message': f'Task {task_id} deleted'}, 204 # 204 No Content for successful deletion

api.add_resource(TaskList, '/tasks')
api.add_resource(Task, '/tasks/<string:task_id>')

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

در این مثال:

• TaskList(Resource): این کلاس برای مدیریت درخواست‌هایی به آدرس /tasks (بدون شناسه خاص) استفاده می‌شود. متد get آن لیستی از تمام وظایف را برمی‌گرداند. متد post برای ایجاد یک وظیفه جدید استفاده می‌شود.
• Task(Resource): این کلاس برای مدیریت درخواست‌هایی به آدرس /tasks/<task_id> (با یک شناسه خاص) استفاده می‌شود. متدهای get، put و delete به ترتیب برای بازیابی، به‌روزرسانی و حذف یک وظیفه خاص استفاده می‌شوند.
• api.add_resource(): این متد یک Resource را به یک مسیر (URL) خاص مرتبط می‌کند. متغیر <string:task_id> در مسیر، به Flask-RESTful می‌گوید که task_id یک پارامتر متغیر از نوع رشته است که از URL استخراج شده و به متدهای کلاس Task پاس داده می‌شود.

حالا می‌توانید برنامه را اجرا کرده و با ابزارهایی مانند Postman یا curl تست کنید:

python your_app_name.py

در مرورگر یا با curl:

GET http://127.0.0.1:5000/tasks
GET http://127.0.0.1:5000/tasks/task1
DELETE http://127.0.0.1:5000/tasks/task2

این مثال پایه، هسته اصلی Flask-RESTful را نشان می‌دهد: تعریف Resourceها و نگاشت آن‌ها به URLها. در بخش‌های بعدی، به سراغ ابزارهای پیشرفته‌تر Flask-RESTful خواهیم رفت.

اعتبارسنجی درخواست‌ها با reqparse: قلب API های مستحکم

یکی از مهمترین جنبه‌های ساخت APIهای قوی، اعتبارسنجی دقیق داده‌های ورودی است. API شما باید بتواند با اطمینان، پارامترهای دریافتی را بررسی کرده و در صورت عدم مطابقت با انتظارات، خطاهای معنی‌داری را برگرداند. Flask-RESTful با ماژول reqparse این فرآیند را به شکل چشمگیری ساده می‌کند. reqparse به شما اجازه می‌دهد تا پارامترهای مورد انتظار را تعریف کنید، اعتبارسنجی‌های مختلفی را بر روی آن‌ها اعمال کنید و به طور خودکار خطاهای 400 Bad Request را در صورت عدم تطابق تولید کنید.

بیایید به سراغ مثال Task Manager برویم و متد post در TaskList و متد put در Task را با استفاده از reqparse کامل کنیم.

تعریف آرگومان‌ها

شما آرگومان‌های مورد انتظار را با ایجاد یک شیء RequestParser و اضافه کردن آرگومان‌ها با متد add_argument تعریف می‌کنید. هر آرگومان می‌تواند دارای ویژگی‌های مختلفی باشد.

from flask import Flask, request
from flask_restful import Api, Resource, reqparse, abort

app = Flask(__name__)
api = Api(app)

tasks = {
    'task1': {'description': 'Write a blog post', 'done': False},
    'task2': {'description': 'Learn Flask-RESTful', 'done': True},
}

# parser برای ایجاد یک تسک جدید
task_post_args = reqparse.RequestParser()
task_post_args.add_argument('description', type=str, required=True, help='Description of the task is required')
task_post_args.add_argument('done', type=bool, required=False, default=False, help='Status of the task')

# parser برای به‌روزرسانی یک تسک
task_put_args = reqparse.RequestParser()
task_put_args.add_argument('description', type=str, help='Description of the task')
task_put_args.add_argument('done', type=bool, help='Status of the task')

# --- کلاس TaskList ---
class TaskList(Resource):
    def get(self):
        return tasks, 200

    def post(self):
        args = task_post_args.parse_args()
        
        # تولید یک task_id جدید (مثلاً بر اساس تعداد تسک‌های موجود)
        new_task_id = f'task{len(tasks) + 1}'
        
        tasks[new_task_id] = {'description': args['description'], 'done': args['done']}
        return {new_task_id: tasks[new_task_id]}, 201

# --- کلاس Task ---
class Task(Resource):
    def get(self, task_id):
        if task_id not in tasks:
            abort(404, message="Task {} doesn't exist".format(task_id))
        return tasks[task_id], 200

    def put(self, task_id):
        if task_id not in tasks:
            abort(404, message="Task {} doesn't exist".format(task_id))
            
        args = task_put_args.parse_args()
        
        # فقط فیلدهایی که در درخواست آمده‌اند را به‌روزرسانی می‌کنیم
        if args['description'] is not None:
            tasks[task_id]['description'] = args['description']
        if args['done'] is not None:
            tasks[task_id]['done'] = args['done']
            
        return tasks[task_id], 200

    def delete(self, task_id):
        if task_id not in tasks:
            abort(404, message="Task {} doesn't exist".format(task_id))
        del tasks[task_id]
        return {'message': f'Task {task_id} deleted'}, 204

api.add_resource(TaskList, '/tasks')
api.add_resource(Task, '/tasks/<string:task_id>')

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

آرگومان‌های الزامی و نوع داده‌ها

در متد add_argument، پارامترهای کلیدی مختلفی وجود دارد که به شما کنترل زیادی بر روی اعتبارسنجی می‌دهند:

• name (الزامی): نام پارامتر در درخواست.
• type: نوع داده مورد انتظار (str, int, float, bool و غیره). اگر داده ورودی قابل تبدیل به این نوع نباشد، reqparse خطا می‌دهد. می‌توانید توابع سفارشی نیز به عنوان type پاس دهید.
• required: یک مقدار بولی (True یا False) که مشخص می‌کند آیا این پارامتر در درخواست الزامی است یا خیر. اگر True باشد و پارامتر در درخواست نباشد، reqparse به طور خودکار خطای 400 برمی‌گرداند.
• help: یک پیام خطا که در صورت عدم وجود پارامتر الزامی یا نامعتبر بودن نوع آن، به مشتری API بازگردانده می‌شود.
• default: مقداری پیش‌فرض که در صورت عدم وجود پارامتر در درخواست، به آن اختصاص داده می‌شود.
• choices: لیستی از مقادیر مجاز. اگر مقدار ورودی در این لیست نباشد، خطا تولید می‌شود.
• action: نحوه هندل کردن پارامترهای تکراری (مثلاً 'append' برای ساخت لیست از مقادیر).
• location: مشخص می‌کند پارامتر باید در کجای درخواست جستجو شود (مثلاً 'json', 'form', 'args' برای query parameters, 'headers', 'cookies'). به طور پیش‌فرض، reqparse در چندین مکان به ترتیب جستجو می‌کند.

در مثال بالا:

• برای description در task_post_args، required=True است، به این معنی که بدون آن درخواست نامعتبر است.
• برای done در task_post_args، required=False و default=False است، یعنی اگر فرستاده نشود، مقدار پیش‌فرض False خواهد بود.
• در task_put_args، هیچ یک از فیلدها required=True نیستند، زیرا هنگام به‌روزرسانی ممکن است فقط یکی از فیلدها ارسال شود.

مکان‌های مختلف درخواست (Headers, JSON Body, Query Params)

reqparse می‌تواند پارامترها را از مکان‌های مختلفی در درخواست HTTP استخراج کند. رایج‌ترین آن‌ها عبارتند از:

• 'args': پارامترهای URL (query string)، مثل ?name=value.
• 'form': داده‌های فرم ارسال شده با Content-Type: application/x-www-form-urlencoded.
• 'json': داده‌های JSON ارسال شده در بدنه درخواست با Content-Type: application/json.
• 'headers': هدرهای درخواست.
• 'cookies': کوکی‌های درخواست.

به طور پیش‌فرض، reqparse به ترتیب زیر به دنبال آرگومان‌ها می‌گردد: `args`, `json`, `form`, `headers`, `cookies` (این ترتیب را می‌توانید با parser.location تغییر دهید). اما می‌توانید با پارامتر location در add_argument، مکان خاصی را مشخص کنید. مثلاً اگر بخواهید یک توکن احراز هویت را از هدر بخوانید:

auth_parser = reqparse.RequestParser()
auth_parser.add_argument('Authorization', type=str, location='headers', required=True, help='Authorization header is required')

# در متد Resource خود
# args = auth_parser.parse_args()
# token = args['Authorization']

استفاده از parse_args برای اعتبارسنجی

پس از تعریف RequestParser و اضافه کردن آرگومان‌ها، شما با فراخوانی متد parse_args() در داخل متد Resource خود، فرآیند اعتبارسنجی را آغاز می‌کنید. این متد:

• تمام آرگومان‌های تعریف شده را از درخواست استخراج می‌کند.
• آن‌ها را بر اساس نوع داده و سایر قوانین اعتبارسنجی می‌کند.
• اگر اعتبارسنجی با موفقیت انجام شود، یک دیکشنری حاوی مقادیر تجزیه شده را برمی‌گرداند.
• اگر هر گونه خطایی در اعتبارسنجی (مثلاً پارامتر الزامی وجود نداشته باشد یا نوع داده اشتباه باشد) رخ دهد، به طور خودکار یک پاسخ 400 Bad Request با جزئیات خطا به مشتری API برمی‌گرداند و اجرای کد در Resource متوقف می‌شود. این ویژگی اتوماتیک، بخش عظیمی از کد هندلینگ خطا را از روی دوش شما برمی‌دارد.

با reqparse، شما یک راهکار قدرتمند و تمیز برای اعتبارسنجی داده‌های ورودی در APIهای Flask-RESTful خود دارید که به حفظ یکپارچگی داده‌ها و افزایش قابلیت اطمینان API کمک شایانی می‌کند.

سریالیزیشن داده‌ها با fields و marshal_with: خروجی‌های استاندارد

یکی دیگر از چالش‌های اصلی در ساخت APIهای RESTful، اطمینان از این است که پاسخ‌های API همیشه فرمت و ساختار یکسانی دارند. این موضوع به ویژه زمانی اهمیت پیدا می‌کند که آبجکت‌های پایتون پیچیده‌ای (مانند مدل‌های ORM) را به JSON تبدیل می‌کنید. Flask-RESTful با ارائه ماژول fields و decorator marshal_with، این فرآیند سریالیزیشن را به شکلی زیبا و کارآمد مدیریت می‌کند.

marshal_with به شما اجازه می‌دهد تا ساختار پاسخ‌های API خود را به وضوح تعریف کنید، اطمینان حاصل کنید که فقط فیلدهای مورد نظر نمایش داده می‌شوند، و حتی داده‌ها را قبل از ارسال قالب‌بندی کنید. این کار به کپسوله‌سازی منطق نمایش داده‌ها کمک می‌کند و API شما را قابل پیش‌بینی‌تر و مصرف‌کننده راضی‌تر می‌کند.

معرفی flask_restful.fields

ماژول flask_restful.fields شامل انواع داده‌های مختلفی است که می‌توانید برای تعریف ساختار خروجی خود استفاده کنید. برخی از رایج‌ترین آن‌ها عبارتند از:

• fields.Raw: نوع پایه، داده را به صورت خام برمی‌گرداند.
• fields.String: یک فیلد رشته‌ای.
• fields.Integer: یک فیلد عددی صحیح.
• fields.Float: یک فیلد عددی اعشاری.
• fields.Boolean: یک فیلد بولی.
• fields.DateTime: یک فیلد تاریخ و زمان، که می‌تواند فرمت‌بندی شود.
• fields.Url: یک فیلد برای ساخت URL، با استفاده از url_for.
• fields.List: برای لیستی از فیلدهای دیگر.
• fields.Nested: برای آبجکت‌های تودرتو (nested objects).

اعمال marshal_with به منابع

برای استفاده از این قابلیت، ابتدا باید یک دیکشنری تعریف کنید که ساختار خروجی مورد نظر شما را با استفاده از انواع fields مشخص می‌کند. سپس، این دیکشنری را به عنوان آرگومان به decorator @marshal_with پاس می‌دهید و آن را بر روی متدهای Resource خود اعمال می‌کنید.

بیایید مثال Task Manager را ادامه دهیم و خروجی‌های get را استانداردسازی کنیم.

from flask import Flask, request, url_for
from flask_restful import Api, Resource, reqparse, abort, fields, marshal_with

app = Flask(__name__)
api = Api(app)

tasks = {
    'task1': {'description': 'Write a blog post', 'done': False},
    'task2': {'description': 'Learn Flask-RESTful', 'done': True},
    'task3': {'description': 'Buy groceries', 'done': False},
}

# parser برای ایجاد یک تسک جدید
task_post_args = reqparse.RequestParser()
task_post_args.add_argument('description', type=str, required=True, help='Description of the task is required')
task_post_args.add_argument('done', type=bool, required=False, default=False, help='Status of the task')

# parser برای به‌روزرسانی یک تسک
task_put_args = reqparse.RequestParser()
task_put_args.add_argument('description', type=str, help='Description of the task')
task_put_args.add_argument('done', type=bool, help='Status of the task')

# تعریف ساختار خروجی برای یک تسک
resource_fields = {
    'description': fields.String,
    'done': fields.Boolean,
    'uri': fields.Url('task_ep', absolute=True) # ساخت یک URL پویا برای هر تسک
}

# --- کلاس TaskList ---
class TaskList(Resource):
    @marshal_with(resource_fields)
    def get(self):
        # برایmarshal_with، باید یک دیکشنری برگردانید که در آن کلیدها نام تسک و مقادیر خود آبجکت‌های تسک باشند
        # یا یک لیست از دیکشنری ها.
        # اینجا یک لیست از آبجکت‌های تسک (به عنوان دیکشنری) را برمی‌گردانیم.
        # برای Url field، باید نام endpoint را مشخص کنید. 'task_ep' نام endpoint کلاس Task است.
        # Flask-RESTful این نام را به طور خودکار بر اساس add_resource(Task, '/tasks/<string:task_id>', endpoint='task_ep') تنظیم می‌کند.
        # اگر endpoint را مشخص نکنید، از نام کلاس در حالت lowercase استفاده می‌کند: task.
        
        # برای هر تسک، باید task_id آن را برای ساخت URL اضافه کنیم.
        # marshal_with انتظار دارد آبجکتی که برمی‌گردد دارای تمام فیلدهای لازم برای ساخت 'uri' باشد.
        # در این حالت، 'task_id' باید در دیکشنری هر تسک موجود باشد.
        
        # بیایید ساختار tasks را تغییر دهیم تا task_id داخل خود آبجکت تسک هم باشد.
        # یا می‌توانیم برای هر آیتم، task_id را به صورت موقت اضافه کنیم.
        
        # راه حل: marshal_with روی لیست دیکشنری‌ها کار می‌کند.
        # باید یک لیست از دیکشنری‌ها برگردانیم که هر دیکشنری یک تسک است و شامل description و done باشد.
        # همچنین برای uri، marshal_with به task_id نیاز دارد.
        # می‌توانیم یک wrapper ایجاد کنیم یا dict را تغییر دهیم.
        
        # برای سادگی، فعلاً فرض می‌کنیم که 'id' هم در آبجکت تسک وجود دارد و uri را به آن متصل می‌کنیم.
        # اما برای نمونه‌ای که داریم، باید آیتم‌ها را قبل از مارشال کردن دستکاری کنیم.
        
        # راه حل بهتر: یک متد جداگانه برای آماده‌سازی داده‌ها برای marshal_with
        prepared_tasks = []
        for task_id, task_data in tasks.items():
            prepared_task = task_data.copy()
            prepared_task['id'] = task_id # Flask-RESTful از این برای ساخت URL استفاده می‌کند.
            prepared_tasks.append(prepared_task)
        return prepared_tasks, 200

    def post(self):
        args = task_post_args.parse_args()
        new_task_id = f'task{len(tasks) + 1}'
        tasks[new_task_id] = {'description': args['description'], 'done': args['done']}
        
        # برای برگرداندن پاسخ با marshal_with، باید آن را روی یک آبجکت برگردانده شده اعمال کنیم.
        # اما اینجا ما یک آبجکت جدید را برمی‌گردانیم، می‌توانیم آن را مستقیماً marshal کنیم.
        # اما برای consistency، بهتر است یک decorator بر روی متد get اعمال شود.
        # برای post، معمولاً آبجکت جدید و کد 201 را برمی‌گردانیم.
        
        # در این حالت خاص، اگر بخواهیم خروجی post هم طبق resource_fields باشد:
        prepared_task = tasks[new_task_id].copy()
        prepared_task['id'] = new_task_id
        # اگر بخواهید خروجی POST هم با resource_fields مارشال شود:
        # return marshal(prepared_task, resource_fields), 201
        # اما معمولاً برای POST فقط آبجکت ایجاد شده را برمی‌گردانند.
        return {new_task_id: tasks[new_task_id]}, 201


# --- کلاس Task ---
class Task(Resource):
    @marshal_with(resource_fields)
    def get(self, task_id):
        if task_id not in tasks:
            abort(404, message="Task {} doesn't exist".format(task_id))
        
        # برای marshal_with، باید 'id' را به آبجکت اضافه کنیم تا UriField بتواند آن را بسازد.
        retrieved_task = tasks[task_id].copy()
        retrieved_task['id'] = task_id
        return retrieved_task, 200

    @marshal_with(resource_fields) # همچنین می‌توانیم برای PUT هم marshal_with اعمال کنیم
    def put(self, task_id):
        if task_id not in tasks:
            abort(404, message="Task {} doesn't exist".format(task_id))
            
        args = task_put_args.parse_args()
        
        if args['description'] is not None:
            tasks[task_id]['description'] = args['description']
        if args['done'] is not None:
            tasks[task_id]['done'] = args['done']
            
        updated_task = tasks[task_id].copy()
        updated_task['id'] = task_id
        return updated_task, 200

    def delete(self, task_id):
        if task_id not in tasks:
            abort(404, message="Task {} doesn't exist".format(task_id))
        del tasks[task_id]
        return {'message': f'Task {task_id} deleted'}, 204

api.add_resource(TaskList, '/tasks', endpoint='tasklist_ep') # نام endpoint برای TaskList
api.add_resource(Task, '/tasks/<string:task_id>', endpoint='task_ep') # نام endpoint برای Task

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

توضیحات مهم:

• resource_fields: یک دیکشنری که ساختار خروجی مورد نظر برای یک تسک را تعریف می‌کند.
• fields.Url('task_ep', absolute=True): این فیلد یک URL کامل برای منبع (تسک) تولید می‌کند. task_ep نام endpoint (که در api.add_resource تعریف می‌شود) برای منبع Task است. absolute=True باعث می‌شود URL کامل با دامنه و پروتکل تولید شود. Flask-RESTful به طور خودکار task_id را از آبجکتی که شما برمی‌گردانید، برای پر کردن URL استفاده می‌کند. به همین دلیل مجبور شدیم 'id' را به دیکشنری‌های تسک اضافه کنیم.
• @marshal_with(resource_fields): این decorator بر روی متدهای get در TaskList و Task و put در Task اعمال شده است. هر آبجکتی که توسط این متدها برگردانده شود، با استفاده از resource_fields سریالیزه خواهد شد. اگر متد یک دیکشنری یا لیست از دیکشنری‌ها را برگرداند، marshal_with آن را به فرمت JSON تبدیل می‌کند.

فیلدهای سفارشی و پیچیده

شما می‌توانید فیلدهای سفارشی خود را با ارث‌بری از fields.Raw ایجاد کنید. این کار به شما امکان می‌دهد تا منطق پیچیده‌تری را برای تبدیل داده‌ها اعمال کنید. مثلاً، فرض کنید می‌خواهیم وضعیت done را به جای True/False به "Complete"/"Pending" تبدیل کنیم:

from flask_restful import fields

class CustomStatusField(fields.Raw):
    def format(self, value):
        return "Complete" if value else "Pending"

resource_fields = {
    'description': fields.String,
    'status': CustomStatusField(attribute='done'), # از attribute برای مشخص کردن فیلد اصلی استفاده می‌کنیم
    'uri': fields.Url('task_ep', absolute=True)
}

در اینجا، CustomStatusField یک فیلد سفارشی است که متد format آن منطق تبدیل را انجام می‌دهد. attribute='done' به CustomStatusField می‌گوید که برای دریافت مقدار، از فیلد done در آبجکت اصلی استفاده کند.

همچنین، fields.Nested برای نمایش آبجکت‌های تودرتو بسیار مفید است. مثلاً اگر یک تسک دارای یک “کاربر” باشد که آن کاربر خود دارای فیلدهای name و email است، می‌توانید ساختار زیر را تعریف کنید:

user_fields = {
    'name': fields.String,
    'email': fields.String
}

task_with_user_fields = {
    'description': fields.String,
    'done': fields.Boolean,
    'user': fields.Nested(user_fields) # کاربر به عنوان یک آبجکت تو در تو
}

با fields و marshal_with، کنترل کاملی بر روی نحوه نمایش داده‌های خود در پاسخ‌های API دارید، که به یکپارچگی، خوانایی و کارآمدی API شما کمک شایانی می‌کند. این ابزار به ویژه برای APIهای بزرگ و پیچیده که با انواع مختلف داده‌ها سروکار دارند، ضروری است.

مدیریت خطاها: پاسخ‌های خوانا و کارآمد

یک API پیشرفته و حرفه‌ای، نه تنها باید بتواند درخواست‌های صحیح را به درستی پردازش کند، بلکه باید در مواجهه با خطاها نیز پاسخ‌های مفید، سازگار و با کد وضعیت HTTP مناسب را برگرداند. Flask-RESTful دارای مکانیزم‌های داخلی برای مدیریت خطاهاست، اما به شما اجازه می‌دهد تا این مکانیزم‌ها را برای برآورده کردن نیازهای خاص خود سفارشی‌سازی کنید.

به طور پیش‌فرض، Flask-RESTful خطاهایی را که از طریق reqparse تولید می‌شوند (مثلاً 400 Bad Request برای پارامترهای ناقص یا نامعتبر) به طور خودکار مدیریت می‌کند. همچنین، اگر متدی در یک Resource یک استثنا را بالا بیاورد، Flask-RESTful آن را به یک پاسخ 500 Internal Server Error تبدیل می‌کند. با این حال، شما اغلب می‌خواهید خطاهای خاصی را با کدهای وضعیت HTTP مشخص (مانند 404 Not Found، 401 Unauthorized، 403 Forbidden) و با پیام‌های خطای سفارشی برگردانید.

خطاهای پیش‌فرض Flask-RESTful

همانطور که قبلاً دیدید، reqparse به طور خودکار خطاهای 400 را هندل می‌کند. تابع abort از Flask-RESTful نیز برای برگرداندن پاسخ‌های خطا با کد وضعیت HTTP دلخواه بسیار مفید است. این تابع باعث می‌شود که درخواست متوقف شده و یک پاسخ خطا به مشتری بازگردانده شود.

from flask_restful import abort

# در متد get یا put از کلاس Task
if task_id not in tasks:
    abort(404, message="Task {} doesn't exist".format(task_id))

abort یک پاسخ JSON با ساختار {'message': 'your error message'} و کد وضعیت HTTP مشخص شده را برمی‌گرداند. این ساده‌ترین راه برای مدیریت خطاهای رایج مانند “منبع یافت نشد” است.

خطاهای سفارشی و هندلرهای استثنا

برای کنترل دقیق‌تر بر روی پاسخ‌های خطا، می‌توانید هندلرهای استثنا (Exception Handlers) سفارشی تعریف کنید. Flask-RESTful به شما اجازه می‌دهد تا توابعی را برای هندل کردن انواع خاصی از استثناها یا کدهای وضعیت HTTP، ثبت کنید. این کار با استفاده از api.handle_error یا api.error_router انجام می‌شود.

بیایید یک استثنای سفارشی تعریف کنیم و سپس یک هندلر برای آن بنویسیم. فرض کنید می‌خواهیم یک خطای “پردازش نامعتبر” (422 Unprocessable Entity) برای زمانی که منطق تجاری ما با مشکل روبرو می‌شود، داشته باشیم.

from flask import Flask, jsonify
from flask_restful import Api, Resource, reqparse, abort, fields, marshal_with

app = Flask(__name__)
api = Api(app)

tasks = {
    'task1': {'description': 'Write a blog post', 'done': False},
    'task2': {'description': 'Learn Flask-RESTful', 'done': True},
    'task3': {'description': 'Buy groceries', 'done': False},
}

# ... (reqparse و resource_fields مانند قبل) ...

# تعریف یک استثنای سفارشی
class InvalidUsage(Exception):
    status_code = 400

    def __init__(self, message, status_code=None, payload=None):
        Exception.__init__(self)
        self.message = message
        if status_code is not None:
            self.status_code = status_code
        self.payload = payload

    def to_dict(self):
        rv = dict(self.payload or ())
        rv['message'] = self.message
        return rv

# هندلر برای استثنای سفارشی
@api.errorhandler(InvalidUsage)
def handle_invalid_usage(error):
    response = jsonify(error.to_dict())
    response.status_code = error.status_code
    return response

# مثال استفاده از خطای سفارشی در یک Resource
class SpecialTask(Resource):
    def post(self):
        # فرض کنید یک شرط خاص باعث خطای منطقی می‌شود
        if request.json and request.json.get('force_error'):
            raise InvalidUsage("This task cannot be processed right now.", status_code=422)
        
        # ... منطق عادی ...
        return {'message': 'Special task created'}, 201

# api.add_resource(SpecialTask, '/special-task')
# api.add_resource(TaskList, '/tasks', endpoint='tasklist_ep')
# api.add_resource(Task, '/tasks/<string:task_id>', endpoint='task_ep')

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

در این مثال:

• InvalidUsage: یک کلاس استثنای سفارشی است که یک message و status_code را می‌پذیرد. متد to_dict آن، خروجی JSON خطا را فرمت‌بندی می‌کند.
• @api.errorhandler(InvalidUsage): این decorator یک تابع را به عنوان هندلر برای استثنای InvalidUsage ثبت می‌کند. هر زمان که این استثنا در هر کجای برنامه Flask-RESTful شما بالا بیاید، این تابع فراخوانی می‌شود.
• در SpecialTask.post، ما به صورت دستی این استثنا را بالا می‌آوریم تا نحوه کارکرد آن را نشان دهیم.

این رویکرد به شما اجازه می‌دهد تا ساختار پاسخ‌های خطا را به طور کامل کنترل کنید و آن‌ها را با معماری RESTful خود سازگار کنید. شما می‌توانید برای کدهای وضعیت HTTP عمومی (مانند 404 یا 500) نیز هندلرهای سفارشی بنویسید تا پیام‌های خطای پیش‌فرض Flask-RESTful را با پیام‌های خود جایگزین کنید.

ساختاردهی پاسخ‌های خطا

یک الگوی رایج برای پاسخ‌های خطا در APIها، استفاده از ساختاری استاندارد است که معمولاً شامل موارد زیر می‌شود:

• code: کد وضعیت HTTP.
• message: یک پیام خوانا برای توسعه‌دهنده.
• errors: (اختیاری) یک دیکشنری یا لیست از جزئیات خطاهای خاص (مثلاً خطاهای اعتبارسنجی فیلدها).

می‌توانید هندلر خطای خود را برای تولید این ساختار تنظیم کنید. برای مثال، برای خطاهای 400 که از reqparse می‌آیند، Flask-RESTful به طور پیش‌فرض یک دیکشنری {'message': {'field_name': 'error message'}} برمی‌گرداند. می‌توانید یک هندلر سراسری برای 400 Bad Request بنویسید تا این خروجی را به فرمت دلخواه شما تبدیل کند.

# مثال هندلر سفارشی برای خطاهای 400 از reqparse
@api.errorhandler(400)
def handle_bad_request(error):
    # error.data شامل دیکشنری خطاهای parse_args است
    return {
        'code': 400,
        'message': 'Invalid input data',
        'errors': error.data.get('errors') if hasattr(error, 'data') else {}
    }, 400

با مدیریت قوی خطاها، API شما نه تنها پایدارتر می‌شود، بلکه تجربه بهتری را برای مصرف‌کنندگان API فراهم می‌کند، زیرا آن‌ها می‌توانند به راحتی مشکلات را شناسایی و برطرف کنند.

احراز هویت و مجوزدهی: امنیت API های پیشرفته

ساخت APIهای پیشرفته بدون در نظر گرفتن امنیت، ناقص خواهد بود. احراز هویت (Authentication) به معنای تایید هویت کاربر است (شما کی هستید؟)، در حالی که مجوزدهی (Authorization) به معنای تعیین دسترسی کاربر به منابع خاص است (به چه چیزهایی دسترسی دارید؟). Flask-RESTful به طور مستقیم مکانیزم‌های احراز هویت یا مجوزدهی را ارائه نمی‌دهد، اما به راحتی با اکستنشن‌های موجود Flask ادغام می‌شود.

رایج‌ترین روش برای احراز هویت در APIهای RESTful، استفاده از توکن‌های مبتنی بر JWT (JSON Web Tokens) است. Flask-JWT-Extended یک اکستنشن عالی برای Flask است که پیاده‌سازی JWT را بسیار ساده می‌کند.

مفهوم توکن‌های JWT

JWT یک استاندارد باز (RFC 7519) است که یک راهکار فشرده و خود-شامل برای انتقال اطلاعات بین طرفین به صورت امن ارائه می‌دهد. این توکن‌ها از سه بخش تشکیل شده‌اند: Header (هدر), Payload (بار داده), و Signature (امضا).

• Header: شامل نوع توکن (JWT) و الگوریتم امضا (مثلاً HMAC SHA256).
• Payload: شامل ادعاهایی (claims) در مورد کاربر و متادیتاهای دیگر است. این اطلاعات رمزگذاری نمی‌شوند، فقط کدگذاری (encoded) می‌شوند، بنابراین اطلاعات حساس را نباید در اینجا قرار داد.
• Signature: برای تایید اینکه توکن توسط فرستنده اصلی ایجاد شده و در طول مسیر تغییر نکرده است، استفاده می‌شود.

زمانی که کاربر با نام کاربری و رمز عبور خود احراز هویت می‌شود، سرور یک JWT به او برمی‌گرداند. کاربر در درخواست‌های بعدی خود، این توکن را معمولاً در هدر Authorization (با پیشوند Bearer) ارسال می‌کند. سرور سپس توکن را اعتبارسنجی کرده و بر اساس اطلاعات داخل آن، هویت و دسترسی کاربر را مشخص می‌کند.

ادغام با Flask-JWT-Extended

ابتدا، Flask-JWT-Extended را نصب کنید:

pip install Flask-JWT-Extended

سپس، آن را در برنامه Flask خود تنظیم کنید:

from flask import Flask, request, jsonify
from flask_restful import Api, Resource, reqparse, abort, fields, marshal_with
from flask_jwt_extended import create_access_token, jwt_required, JWTManager, get_jwt_identity, get_jwt

app = Flask(__name__)
api = Api(app)

app.config["JWT_SECRET_KEY"] = "super-secret-key-that-should-be-in-env-vars" # کلید مخفی برای امضای توکن‌ها
jwt = JWTManager(app)

# یک دیکشنری ساده برای شبیه‌سازی کاربران
users = {
    'john': {'password': 'password123', 'roles': ['user']},
    'admin': {'password': 'adminpass', 'roles': ['admin', 'user']},
}

# ... (reqparse و resource_fields مانند قبل) ...

# Resource برای لاگین کردن و دریافت توکن
class Login(Resource):
    def post(self):
        username = request.json.get('username', None)
        password = request.json.get('password', None)

        if not username or not password or username not in users or users[username]['password'] != password:
            return {'msg': 'Bad username or password'}, 401

        # ایجاد توکن دسترسی
        access_token = create_access_token(identity=username, additional_claims={"roles": users[username]['roles']})
        return jsonify(access_token=access_token)

# Resource برای نمایش یک تسک که نیاز به احراز هویت دارد
class ProtectedTask(Resource):
    @jwt_required() # این decorator تضمین می‌کند که این endpoint فقط با توکن معتبر قابل دسترسی است
    @marshal_with(resource_fields)
    def get(self, task_id):
        current_user = get_jwt_identity() # نام کاربری از توکن
        user_roles = get_jwt()['roles'] # نقش‌های کاربر از additional_claims

        if task_id not in tasks:
            abort(404, message="Task {} doesn't exist".format(task_id))
            
        # فقط ادمین‌ها می‌توانند تسک‌های خاص را ببینند (مثلاً 'task1' فقط برای ادمین)
        if task_id == 'task1' and 'admin' not in user_roles:
            abort(403, message="Access to task {} forbidden".format(task_id))

        retrieved_task = tasks[task_id].copy()
        retrieved_task['id'] = task_id
        return retrieved_task, 200

api.add_resource(Login, '/login')
api.add_resource(ProtectedTask, '/protected-tasks/<string:task_id>', endpoint='protected_task_ep')

# ... (سایر api.add_resource ها) ...

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

محافظت از Endpoints

با استفاده از decorator @jwt_required()، می‌توانید هر متد Resource را محافظت کنید. زمانی که یک درخواست به این متد می‌رسد:

• اگر هدر Authorization وجود نداشته باشد یا توکن نامعتبر باشد، Flask-JWT-Extended به طور خودکار یک پاسخ 401 Unauthorized برمی‌گرداند.
• اگر توکن معتبر باشد، درخواست به متد Resource شما ادامه پیدا می‌کند.

در داخل متد محافظت شده، می‌توانید از توابع کمکی Flask-JWT-Extended مانند get_jwt_identity() برای دریافت هویت کاربر (در اینجا نام کاربری) و get_jwt() برای دریافت کل Payload توکن (از جمله additional_claims که در زمان ایجاد توکن اضافه کرده‌اید) استفاده کنید.

پیاده‌سازی مجوزدهی نقش‌محور (Role-Based Access Control)

برای پیاده‌سازی مجوزدهی (RBAC)، می‌توانید نقش‌های کاربر را به عنوان additional_claims در JWT اضافه کنید. سپس در متدهای Resource محافظت شده خود، این نقش‌ها را بررسی کرده و بر اساس آن‌ها تصمیم بگیرید که آیا کاربر مجاز به انجام عملیات خاصی هست یا خیر.

در مثال ProtectedTask:

user_roles = get_jwt()['roles'] # نقش‌ها را از توکن می‌خوانیم
if task_id == 'task1' and 'admin' not in user_roles:
    abort(403, message="Access to task {} forbidden".format(task_id))

این قطعه کد بررسی می‌کند که اگر کاربر قصد دسترسی به task1 را دارد و نقش admin را ندارد، یک خطای 403 Forbidden برگردانده شود. این یک الگوی اساسی برای مجوزدهی است که می‌توانید آن را پیچیده‌تر کنید (مثلاً با decoratorهای سفارشی برای نقش‌ها).

با ادغام Flask-JWT-Extended، می‌توانید یک لایه امنیتی قوی و مقیاس‌پذیر به APIهای Flask-RESTful خود اضافه کنید که برای کاربردهای تولیدی ضروری است. همیشه به یاد داشته باشید که کلید مخفی JWT را در متغیرهای محیطی نگهداری کنید و هرگز آن را در کد منبع هاردکد نکنید.

صفحه‌بندی (Pagination) و فیلترینگ (Filtering): کار با مجموعه داده‌های بزرگ

هنگامی که API شما با مجموعه داده‌های بزرگی سروکار دارد، برگرداندن تمام داده‌ها در یک درخواست واحد هم ناکارآمد است و هم می‌تواند باعث مشکلات عملکردی شود. برای حل این مشکل، تکنیک‌های صفحه‌بندی (Pagination) و فیلترینگ (Filtering) استفاده می‌شوند. صفحه‌بندی به مشتری API اجازه می‌دهد تا داده‌ها را به صورت تکه‌های کوچک دریافت کند، در حالی که فیلترینگ به او امکان می‌دهد تا فقط زیرمجموعه‌ای از داده‌ها را که معیارهای خاصی را برآورده می‌کنند، درخواست کند.

Flask-RESTful به طور مستقیم توابع صفحه‌بندی و فیلترینگ را ارائه نمی‌دهد، اما می‌توان آن‌ها را به راحتی با استفاده از reqparse برای دریافت پارامترهای درخواست و منطق پایتون برای پردازش داده‌ها پیاده‌سازی کرد.

پیاده‌سازی Pagination بر اساس Offset/Limit

یکی از رایج‌ترین روش‌های صفحه‌بندی، استفاده از پارامترهای offset (شروع از کدام آیتم) و limit (چند آیتم برگردانده شود) در Query String است. Flask-RESTful و reqparse برای مدیریت این پارامترها بسیار مناسب هستند.

from flask import Flask, request
from flask_restful import Api, Resource, reqparse, abort, fields, marshal_with
from flask_jwt_extended import JWTManager # فرض می‌کنیم JWT فعال است

app = Flask(__name__)
api = Api(app)
app.config["JWT_SECRET_KEY"] = "super-secret-key-that-should-be-in-env-vars"
jwt = JWTManager(app)

# ... (users, tasks, resource_fields) ...

# Parser برای پارامترهای صفحه‌بندی
pagination_parser = reqparse.RequestParser()
pagination_parser.add_argument('offset', type=int, default=0, help='Offset for pagination', location='args')
pagination_parser.add_argument('limit', type=int, default=10, help='Limit for pagination', location='args')

class PagedTaskList(Resource):
    @marshal_with(resource_fields)
    def get(self):
        args = pagination_parser.parse_args()
        offset = args['offset']
        limit = args['limit']

        # تبدیل دیکشنری tasks به لیست برای اعمال صفحه‌بندی
        all_tasks_list = []
        for task_id, task_data in tasks.items():
            prepared_task = task_data.copy()
            prepared_task['id'] = task_id
            all_tasks_list.append(prepared_task)

        # اعمال صفحه‌بندی
        paged_tasks = all_tasks_list[offset:offset + limit]

        # ممکن است بخواهید اطلاعات مربوط به صفحه‌بندی (مثل تعداد کل آیتم‌ها) را هم برگردانید.
        # این به معنای عدم استفاده مستقیم از marshal_with بر روی لیست، بلکه برگرداندن یک دیکشنری حاوی لیست و متادیتا است.
        # برای سادگی، فعلا فقط لیست را برمی‌گردانیم.
        
        return paged_tasks, 200

api.add_resource(PagedTaskList, '/paged-tasks', endpoint='paged_tasks_ep')

# ... (سایر api.add_resource ها) ...

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

در این مثال:

• pagination_parser: یک RequestParser جدید برای دریافت offset و limit از Query String (location='args') تعریف شده است. مقادیر پیش‌فرض نیز برای آن‌ها تنظیم شده‌اند.
• در متد get از PagedTaskList، ابتدا پارامترهای صفحه‌بندی را با parse_args() دریافت می‌کنیم.
• سپس، لیست وظایف را بر اساس offset و limit برش می‌دهیم.

برای فراخوانی این API:

GET http://127.0.0.1:5000/paged-tasks?offset=0&limit=2
GET http://127.0.0.1:5000/paged-tasks?offset=1&limit=1

توجه داشته باشید که اگر بخواهید متادیتای صفحه‌بندی (مانند total_items، total_pages، next_page_url و prev_page_url) را نیز برگردانید، باید ساختار resource_fields را تغییر دهید یا یک دیکشنری پیچیده‌تر را برگردانید که در آن لیست واقعی آیتم‌ها زیر یک کلید مانند 'data' قرار گیرد و متادیتا در کنار آن باشد. در این صورت، @marshal_with را بر روی یک فیلد Nested که شامل List آیتم‌هاست، اعمال خواهید کرد.

پیاده‌سازی فیلترینگ (Filtering) بر اساس Query Parameters

فیلترینگ نیز مشابه صفحه‌بندی، از طریق پارامترهای Query String پیاده‌سازی می‌شود. شما می‌توانید پارامترهایی مانند status، category، start_date و غیره را بپذیرید.

# Parser برای پارامترهای فیلترینگ
filtering_parser = reqparse.RequestParser()
filtering_parser.add_argument('done', type=bool, help='Filter by task status', location='args')
filtering_parser.add_argument('description_contains', type=str, help='Filter by description content', location='args')

class FilteredTaskList(Resource):
    @marshal_with(resource_fields)
    def get(self):
        args = filtering_parser.parse_args()
        
        # ترکیب صفحه‌بندی و فیلترینگ
        # args_pagination = pagination_parser.parse_args() # اگر بخواهید صفحه‌بندی هم داشته باشید
        # offset = args_pagination['offset']
        # limit = args_pagination['limit']

        filtered_tasks_list = []
        for task_id, task_data in tasks.items():
            match = True
            
            # فیلتر بر اساس 'done'
            if args['done'] is not None and task_data['done'] != args['done']:
                match = False
            
            # فیلتر بر اساس 'description_contains'
            if args['description_contains'] and args['description_contains'].lower() not in task_data['description'].lower():
                match = False

            if match:
                prepared_task = task_data.copy()
                prepared_task['id'] = task_id
                filtered_tasks_list.append(prepared_task)

        # در صورت تمایل، در اینجا صفحه‌بندی را روی filtered_tasks_list اعمال کنید
        # paged_filtered_tasks = filtered_tasks_list[offset:offset + limit]
        
        return filtered_tasks_list, 200

api.add_resource(FilteredTaskList, '/filtered-tasks', endpoint='filtered_tasks_ep')

# ... (سایر api.add_resource ها) ...

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

در این مثال:

• filtering_parser: پارامترهای done و description_contains را برای فیلترینگ تعریف می‌کند.
• در متد get از FilteredTaskList، ابتدا پارامترهای فیلترینگ را دریافت می‌کنیم.
• سپس، بر روی لیست کامل وظایف حلقه می‌زنیم و فقط آن‌هایی را که با معیارهای فیلتر مطابقت دارند، اضافه می‌کنیم.

برای فراخوانی این API:

GET http://127.0.0.1:5000/filtered-tasks?done=true
GET http://127.0.0.1:5000/filtered-tasks?description_contains=learn
GET http://127.0.0.1:5000/filtered-tasks?done=false&description_contains=write

هنگام کار با دیتابیس (مانند SQLAlchemy)، این عملیات فیلترینگ و صفحه‌بندی به جای انجام در حافظه (مانند مثال‌های بالا)، مستقیماً بر روی کوئری‌های دیتابیس اعمال می‌شوند تا عملکرد بهتری داشته باشند.

با ترکیب reqparse برای گرفتن پارامترها و منطق پایتون برای پردازش داده‌ها، می‌توانید راهکارهای صفحه‌بندی و فیلترینگ قدرتمندی را در APIهای Flask-RESTful خود پیاده‌سازی کنید که برای کار با مجموعه داده‌های بزرگ ضروری هستند.

تست API ها: اطمینان از صحت عملکرد

تست‌نویسی یک جزء حیاتی در چرخه عمر توسعه نرم‌افزار است، به ویژه برای APIها که اغلب توسط سیستم‌های دیگر مصرف می‌شوند. اطمینان از اینکه API شما به درستی کار می‌کند، در مواجهه با ورودی‌های مختلف، و اینکه تغییرات جدید باعث شکست قابلیت‌های موجود نمی‌شوند، اهمیت بالایی دارد. Flask-RESTful، به عنوان یک اکستنشن Flask، به خوبی با قابلیت‌های تست داخلی Flask ادغام می‌شود.

Flask یک test_client داخلی را فراهم می‌کند که به شما امکان می‌دهد درخواست‌های HTTP را به برنامه خود ارسال کرده و پاسخ‌ها را بدون نیاز به اجرای واقعی سرور، بررسی کنید. این ابزار برای نوشتن تست‌های واحد (Unit Tests) و تست‌های یکپارچه‌سازی (Integration Tests) برای API شما بسیار مناسب است.

استفاده از Flask Test Client

برای شروع، شما باید یک فایل تست (مثلاً test_api.py) ایجاد کنید و کلاس تست خود را در آن بنویسید. نیاز دارید که نمونه‌ای از برنامه Flask خود را برای تست ایجاد کنید و سپس test_client آن را بدست آورید.

import unittest
import json
from your_app_name import app, api, tasks, users # فرض می‌کنیم برنامه اصلی شما در your_app_name.py است

# تنظیمات برای تست (مثلاً JWT_SECRET_KEY)
app.config["TESTING"] = True
app.config["JWT_SECRET_KEY"] = "test-secret-key"

class TaskAPITestCase(unittest.TestCase):
    def setUp(self):
        # این متد قبل از هر تست اجرا می‌شود
        self.app = app.test_client()
        self.app.testing = True
        
        # برای هر تست، داده‌ها را به حالت اولیه برگردانید تا تست‌ها مستقل باشند
        self.initial_tasks = {
            'task1': {'description': 'Write a blog post', 'done': False},
            'task2': {'description': 'Learn Flask-RESTful', 'done': True},
            'task3': {'description': 'Buy groceries', 'done': False},
        }
        tasks.clear() # دیکشنری tasks را پاک کنید
        tasks.update(self.initial_tasks) # با داده‌های اولیه پر کنید

    def test_get_all_tasks(self):
        response = self.app.get('/tasks')
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.data)
        self.assertEqual(len(data), len(self.initial_tasks))
        self.assertIn('description', data[0])
        self.assertIn('uri', data[0]) # بررسی فیلد uri که توسط marshal_with اضافه می‌شود

    def test_get_single_task(self):
        response = self.app.get('/tasks/task1')
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.data)
        self.assertEqual(data['description'], 'Write a blog post')
        self.assertEqual(data['done'], False)
        self.assertIn('uri', data)

    def test_get_nonexistent_task(self):
        response = self.app.get('/tasks/task99')
        self.assertEqual(response.status_code, 404)
        data = json.loads(response.data)
        self.assertIn('message', data)
        self.assertEqual(data['message'], "Task task99 doesn't exist")

    def test_create_task_success(self):
        new_task_data = {'description': 'Test new task', 'done': True}
        response = self.app.post('/tasks', data=json.dumps(new_task_data), content_type='application/json')
        self.assertEqual(response.status_code, 201)
        data = json.loads(response.data)
        self.assertIn('task4', data)
        self.assertEqual(data['task4']['description'], 'Test new task')
        self.assertEqual(len(tasks), len(self.initial_tasks) + 1) # بررسی اینکه تسک اضافه شده است

    def test_create_task_missing_description(self):
        new_task_data = {'done': False} # description الزامی است
        response = self.app.post('/tasks', data=json.dumps(new_task_data), content_type='application/json')
        self.assertEqual(response.status_code, 400)
        data = json.loads(response.data)
        self.assertIn('message', data)
        self.assertIn('description', data['message']) # پیام خطای reqparse

    def test_update_task_success(self):
        update_data = {'description': 'Updated description'}
        response = self.app.put('/tasks/task1', data=json.dumps(update_data), content_type='application/json')
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.data)
        self.assertEqual(data['description'], 'Updated description')
        self.assertEqual(tasks['task1']['description'], 'Updated description') # بررسی تغییر در داده‌های اصلی

    def test_delete_task_success(self):
        response = self.app.delete('/tasks/task1')
        self.assertEqual(response.status_code, 204)
        self.assertNotIn('task1', tasks) # بررسی حذف تسک

    def test_login_and_access_protected_task(self):
        # ابتدا لاگین می‌کنیم تا توکن بگیریم
        login_data = {'username': 'john', 'password': 'password123'}
        login_response = self.app.post('/login', data=json.dumps(login_data), content_type='application/json')
        self.assertEqual(login_response.status_code, 200)
        access_token = json.loads(login_response.data)['access_token']

        # با توکن به ProtectedTask دسترسی پیدا می‌کنیم
        headers = {'Authorization': f'Bearer {access_token}'}
        response = self.app.get('/protected-tasks/task2', headers=headers)
        self.assertEqual(response.status_code, 200)
        data = json.loads(response.data)
        self.assertEqual(data['description'], 'Learn Flask-RESTful')

    def test_access_protected_task_no_token(self):
        response = self.app.get('/protected-tasks/task2')
        self.assertEqual(response.status_code, 401)
        data = json.loads(response.data)
        self.assertEqual(data['msg'], 'Missing Authorization Header')
        
    def test_access_forbidden_task_as_user(self):
        login_data = {'username': 'john', 'password': 'password123'}
        login_response = self.app.post('/login', data=json.dumps(login_data), content_type='application/json')
        access_token = json.loads(login_response.data)['access_token']
        headers = {'Authorization': f'Bearer {access_token}'}

        response = self.app.get('/protected-tasks/task1', headers=headers) # task1 برای ادمین‌هاست
        self.assertEqual(response.status_code, 403)
        data = json.loads(response.data)
        self.assertEqual(data['message'], 'Access to task task1 forbidden')


if __name__ == '__main__':
    unittest.main()

برای اجرای تست‌ها:

python -m unittest test_api.py

نوشتن تست‌های واحد (Unit Tests) و یکپارچه‌سازی (Integration Tests)

تست‌های بالا ترکیبی از تست‌های واحد و یکپارچه‌سازی هستند:

• تست‌های واحد: بر روی بخش‌های کوچک و منفرد کد (مانند یک متد خاص در Resource) تمرکز می‌کنند و وابستگی‌ها را با Mock کردن از بین می‌برند. در مثال بالا، تست‌هایی که فقط یک endpoint خاص را بررسی می‌کنند، می‌توانند به عنوان واحد تلقی شوند.
• تست‌های یکپارچه‌سازی: بررسی می‌کنند که چگونه اجزای مختلف سیستم (مانند Flask-RESTful، JWT، منطق تجاری و حتی دیتابیس) با هم کار می‌کنند. تست‌هایی که شامل لاگین و سپس دسترسی به endpoint محافظت شده هستند، تست‌های یکپارچه‌سازی محسوب می‌شوند.

در setUp، مهم است که وضعیت برنامه (مانند دیتابیس یا دیکشنری tasks) را قبل از هر تست به حالت اولیه بازگردانید تا تست‌ها مستقل از یکدیگر اجرا شوند. این کار به جلوگیری از “تداخل” تست‌ها با یکدیگر کمک می‌کند. برای برنامه‌هایی که از دیتابیس واقعی استفاده می‌کنند، معمولاً یک دیتابیس تست جداگانه پیکربندی می‌شود یا از تراکنش‌ها برای Rollback کردن تغییرات پس از هر تست استفاده می‌شود.

با نوشتن مجموعه‌ای جامع از تست‌ها، می‌توانید با اطمینان خاطر بیشتری کد خود را تغییر دهید و توسعه دهید، با علم به اینکه API شما همانطور که انتظار می‌رود عمل می‌کند و در برابر رگرسیون (regression) محافظت می‌شود.

بهترین شیوه‌ها و ملاحظات پیشرفته

ساخت یک API صرفاً به کارکرد آن محدود نمی‌شود؛ یک API پیشرفته باید قابل نگهداری، مقیاس‌پذیر، قابل درک و امن باشد. در ادامه به برخی از بهترین شیوه‌ها و ملاحظات پیشرفته برای APIهای Flask-RESTful اشاره می‌کنیم.

نسخه‌بندی (Versioning) API

APIهای شما با گذشت زمان تکامل می‌یابند. اضافه کردن قابلیت‌های جدید، تغییر ساختار پاسخ‌ها یا حذف فیلدهای قدیمی می‌تواند مشتریان فعلی API شما را با مشکل مواجه کند. برای جلوگیری از این مشکل، نسخه‌بندی API یک راهکار ضروری است.

روش‌های رایج برای نسخه‌بندی API عبارتند از:

1. نسخه‌بندی در URL (URI Versioning): رایج‌ترین و ساده‌ترین روش. نسخه API به عنوان بخشی از مسیر URL مشخص می‌شود.

   
           /api/v1/tasks
           /api/v2/tasks
           

   در Flask-RESTful، این کار با تعریف منابع جداگانه برای هر نسخه و افزودن آن‌ها به api با مسیرهای متفاوت انجام می‌شود:

   
           class TaskV1(Resource):
               # ...
           class TaskV2(Resource):
               # ...
           api.add_resource(TaskV1, '/api/v1/tasks')
           api.add_resource(TaskV2, '/api/v2/tasks')
           

2. نسخه‌بندی در هدر (Header Versioning): نسخه API در یک هدر سفارشی درخواست (مثلاً X-API-Version: 1) یا از طریق هدر Accept (Media Type Versioning) مشخص می‌شود.

   
           Accept: application/vnd.yourapi.v1+json
           

   این روش نیاز به منطق سفارشی برای بررسی هدر در Resourceها یا استفاده از RequestParser دارد.

نسخه‌بندی URL اغلب به دلیل سادگی و خوانایی، ترجیح داده می‌شود.

مستندسازی (OpenAPI/Swagger)

یک API بدون مستندات خوب، مانند یک گنجینه پنهان است. توسعه‌دهندگانی که می‌خواهند از API شما استفاده کنند، به مستندات واضح و به‌روز نیاز دارند. OpenAPI (که قبلاً Swagger نامیده می‌شد) یک استاندارد صنعتی برای تعریف رابط‌های API است.

اکستنشن‌های Flask مانند Flask-Marshmallow-OpenAPI یا Flask-RestX (یک فورک از Flask-RESTPlus که خود بر پایه Flask-RESTful است) به شما کمک می‌کنند تا مستندات OpenAPI را به طور خودکار از کد Flask-RESTful خود تولید کنید. این ابزارها با استفاده از دکوراتورها و کامنت‌های داکسترینگ، اطلاعات لازم را برای تولید مستندات Swagger UI و Swagger Editor فراهم می‌کنند.

داشتن مستندات خودکار، نگهداری آن را آسان‌تر کرده و اطمینان می‌دهد که همیشه با کد شما همگام است.

ملاحظات استقرار (Deployment)

هنگامی که API شما آماده تولید است، باید آن را در یک محیط سرور مستقر کنید. Flask، به عنوان یک فریم‌ورک WSGI، به یک سرور WSGI مانند Gunicorn یا uWSGI نیاز دارد تا درخواست‌ها را مدیریت کند. این سرور WSGI می‌تواند پشت یک پروکسی معکوس (Reverse Proxy) مانند Nginx یا Apache قرار گیرد.

نکات کلیدی برای استقرار:

• استفاده از سرور WSGI: هرگز از سرور توسعه داخلی Flask (app.run(debug=True)) در محیط تولید استفاده نکنید.
• پروکسی معکوس (Nginx/Apache): Nginx برای مدیریت ترافیک، SSL/TLS، لود بالانسینگ و سرویس‌دهی فایل‌های استاتیک عالی است.
• متغیرهای محیطی: اطلاعات حساس مانند JWT_SECRET_KEY، رمزهای عبور دیتابیس و سایر پیکربندی‌های محیطی را از طریق متغیرهای محیطی به برنامه خود پاس دهید و از هاردکد کردن آن‌ها در کد خودداری کنید.
• لاگ‌برداری: لاگ‌های جامع برای نظارت بر عملکرد API و تشخیص مشکلات ضروری هستند.
• نظارت و هشدار: ابزارهای نظارتی (مانند Prometheus, Grafana) برای پیگیری معیارهای عملکرد و تنظیم هشدارها برای مشکلات احتمالی، حیاتی هستند.
• امنیت: اطمینان از به روز بودن تمام وابستگی‌ها، استفاده از HTTPS و اعمال فایروال‌های مناسب.

با رعایت این بهترین شیوه‌ها و ملاحظات پیشرفته، APIهای Flask-RESTful شما نه تنها از نظر فنی قوی خواهند بود، بلکه از نظر قابلیت نگهداری، مقیاس‌پذیری و امنیتی نیز استانداردهای بالایی را رعایت خواهند کرد.

نتیجه‌گیری

Flask-RESTful یک اکستنشن قدرتمند و بسیار مفید برای فریم‌ورک Flask است که توسعه APIهای RESTful را به یک تجربه کارآمد و لذت‌بخش تبدیل می‌کند. از لحظه نصب و راه‌اندازی اولیه تا پیاده‌سازی قابلیت‌های پیشرفته‌ای مانند اعتبارسنجی دقیق درخواست‌ها با reqparse، استانداردسازی خروجی‌ها با fields و marshal_with، مدیریت جامع خطاها، و تامین امنیت با احراز هویت مبتنی بر JWT، Flask-RESTful ابزارهای لازم را در اختیار توسعه‌دهندگان قرار می‌دهد تا APIهایی با کیفیت بالا و قابل اعتماد بسازند.

ما در این پست به تفصیل به جنبه‌های مختلفی پرداختیم که برای ساخت یک API پیشرفته ضروری هستند: از مفاهیم پایه‌ای Resource و API گرفته تا موضوعات پیچیده‌تری مانند صفحه‌بندی، فیلترینگ، تست‌نویسی و بهترین شیوه‌های توسعه و استقرار. با درک عمیق این مفاهیم و استفاده از ابزارهایی که Flask-RESTful ارائه می‌دهد، شما می‌توانید APIهایی بسازید که نه تنها نیازهای فعلی پروژه شما را برآورده می‌کنند، بلکه برای رشد و تکامل در آینده نیز آماده هستند.

به یاد داشته باشید که موفقیت یک API تنها به کدنویسی آن بستگی ندارد، بلکه طراحی خوب، مستندات کامل، و توجه به تجربه مصرف‌کننده (Developer Experience) نیز از اهمیت بالایی برخوردارند. با به‌کارگیری دانش کسب‌شده در این مقاله، شما گام‌های محکمی در جهت تبدیل شدن به یک توسعه‌دهنده API کارآمد با Flask-RESTful برداشته‌اید. اکنون زمان آن است که دانش خود را به عمل تبدیل کرده و APIهای پیشرفته و کاربردی خود را بسازید.