API با Flask — کتابخانه مرجع پایتون

🚀 ساخت API با Flask

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

پیش‌نیاز: آشنایی با Python، مفاهیم API و JSON.

🔹 نصب و import

# نصب (در محیط واقعی)
# pip install flask
# pip install flask-cors  # برای مدیریت CORS

from flask import Flask, request, jsonify, abort
import json
import os

print("Flask is ready!")

🔹 Hello World API

from flask import Flask, jsonify

# ایجاد اپلیکیشن Flask
app = Flask(__name__)

# ===== مسیر اصلی =====
@app.route('/')
def home():
    return jsonify({
        'message': 'Hello, World!',
        'status': 'success'
    })

# ===== مسیر سلام =====
@app.route('/hello/<name>')
def hello_name(name):
    return jsonify({
        'message': f'Hello, {name}!',
        'name': name,
        'status': 'success'
    })

# ===== اجرا =====
if __name__ == '__main__':
    # app.run(debug=True, host='0.0.0.0', port=5000)
    print("Flask app is ready!")
    print("Routes:")
    print("  GET /")
    print("  GET /hello/<name>")

🔹 دریافت داده از درخواست

from flask import Flask, request, jsonify

app = Flask(__name__)

# ===== GET با پارامترها (query string) =====
@app.route('/api/search')
def search():
    # دریافت پارامتر از URL: /api/search?q=python&limit=10
    query = request.args.get('q', default='', type=str)
    limit = request.args.get('limit', default=10, type=int)
    
    return jsonify({
        'query': query,
        'limit': limit,
        'results': [f"Result {i}" for i in range(limit)],
        'status': 'success'
    })

# ===== POST با JSON (بدنه درخواست) =====
@app.route('/api/users', methods=['POST'])
def create_user():
    # دریافت داده JSON از بدنه درخواست
    data = request.get_json()
    
    if not data:
        return jsonify({'error': 'No JSON data provided'}), 400
    
    # اعتبارسنجی ساده
    if 'name' not in data or 'email' not in data:
        return jsonify({'error': 'Name and email are required'}), 400
    
    # ایجاد کاربر (شبیه‌سازی)
    user = {
        'id': 123,
        'name': data['name'],
        'email': data['email'],
        'status': 'created'
    }
    
    return jsonify(user), 201

print("Data handling routes are ready!")
print("  GET /api/search?q=<query>&limit=<limit>")
print("  POST /api/users (with JSON body)")

🔹 مدیریت خطا و وضعیت‌ها

from flask import Flask, jsonify, abort
import json

app = Flask(__name__)

# ===== داده نمونه =====
users = {
    1: {'id': 1, 'name': 'Saba', 'email': '[email protected]'},
    2: {'id': 2, 'name': 'Ali', 'email': '[email protected]'},
    3: {'id': 3, 'name': 'Reza', 'email': '[email protected]'}
}

# ===== دریافت کاربر با ID =====
@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    user = users.get(user_id)
    if not user:
        # بازگرداندن خطای 404 با پیام
        return jsonify({'error': f'User {user_id} not found'}), 404
    return jsonify(user)

# ===== حذف کاربر =====
@app.route('/api/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    if user_id not in users:
        abort(404, description=f'User {user_id} not found')
    
    deleted_user = users.pop(user_id)
    return jsonify({
        'message': 'User deleted successfully',
        'deleted_user': deleted_user
    })

# ===== مدیریت خطای 404 سفارشی =====
@app.errorhandler(404)
def not_found(error):
    return jsonify({
        'error': 'Resource not found',
        'message': str(error)
    }), 404

# ===== مدیریت خطای 400 =====
@app.errorhandler(400)
def bad_request(error):
    return jsonify({
        'error': 'Bad request',
        'message': str(error)
    }), 400

print("Error handling routes are ready!")
print("  GET /api/users/<user_id>")
print("  DELETE /api/users/<user_id>")

🔹 API کامل CRUD

from flask import Flask, request, jsonify

app = Flask(__name__)

# ===== داده در حافظه =====
users = []
next_id = 1

class UserAPI:
    """کلاس مدیریت کاربران با Flask"""
    
    @staticmethod
    def get_all():
        """GET /api/users - دریافت همه کاربران"""
        return jsonify({
            'users': users,
            'count': len(users)
        })
    
    @staticmethod
    def get_one(user_id):
        """GET /api/users/<id> - دریافت یک کاربر"""
        user = next((u for u in users if u['id'] == user_id), None)
        if not user:
            return jsonify({'error': 'User not found'}), 404
        return jsonify(user)
    
    @staticmethod
    def create():
        """POST /api/users - ایجاد کاربر جدید"""
        global next_id
        data = request.get_json()
        
        if not data:
            return jsonify({'error': 'No data provided'}), 400
        
        # اعتبارسنجی
        required = ['name', 'email']
        for field in required:
            if field not in data:
                return jsonify({'error': f'Missing field: {field}'}), 400
        
        user = {
            'id': next_id,
            'name': data['name'],
            'email': data['email'],
            'age': data.get('age', 0)
        }
        users.append(user)
        next_id += 1
        
        return jsonify(user), 201
    
    @staticmethod
    def update(user_id):
        """PUT /api/users/<id> - به‌روزرسانی کاربر"""
        user = next((u for u in users if u['id'] == user_id), None)
        if not user:
            return jsonify({'error': 'User not found'}), 404
        
        data = request.get_json()
        if not data:
            return jsonify({'error': 'No data provided'}), 400
        
        # به‌روزرسانی فیلدها
        user['name'] = data.get('name', user['name'])
        user['email'] = data.get('email', user['email'])
        user['age'] = data.get('age', user.get('age', 0))
        
        return jsonify(user)
    
    @staticmethod
    def delete(user_id):
        """DELETE /api/users/<id> - حذف کاربر"""
        global users
        user = next((u for u in users if u['id'] == user_id), None)
        if not user:
            return jsonify({'error': 'User not found'}), 404
        
        users = [u for u in users if u['id'] != user_id]
        return jsonify({'message': 'User deleted successfully'})

print("CRUD API is ready!")
print("  GET    /api/users")
print("  GET    /api/users/<id>")
print("  POST   /api/users")
print("  PUT    /api/users/<id>")
print("  DELETE /api/users/<id>")

🔹 ساختار API با Blueprint

from flask import Flask, Blueprint, request, jsonify

# ===== ایجاد Blueprint =====
api_bp = Blueprint('api', __name__, url_prefix='/api/v1')

# ===== داده نمونه =====
products = [
    {'id': 1, 'name': 'Laptop', 'price': 1200},
    {'id': 2, 'name': 'Phone', 'price': 800},
    {'id': 3, 'name': 'Tablet', 'price': 500}
]

# ===== مسیرهای Blueprint =====
@api_bp.route('/products')
def get_products():
    return jsonify({'products': products, 'count': len(products)})

@api_bp.route('/products/<int:product_id>')
def get_product(product_id):
    product = next((p for p in products if p['id'] == product_id), None)
    if not product:
        return jsonify({'error': 'Product not found'}), 404
    return jsonify(product)

@api_bp.route('/products', methods=['POST'])
def create_product():
    data = request.get_json()
    if not data or 'name' not in data or 'price' not in data:
        return jsonify({'error': 'Name and price are required'}), 400
    
    new_id = max([p['id'] for p in products]) + 1
    product = {
        'id': new_id,
        'name': data['name'],
        'price': data['price']
    }
    products.append(product)
    return jsonify(product), 201

# ===== ثبت Blueprint در اپلیکیشن =====
app = Flask(__name__)
app.register_blueprint(api_bp)

print("Blueprint API structure is ready!")
print("  GET    /api/v1/products")
print("  GET    /api/v1/products/<id>")
print("  POST   /api/v1/products")

🔹 CORS و Middleware

from flask import Flask, request, jsonify
from flask_cors import CORS

app = Flask(__name__)

# ===== فعال‌سازی CORS =====
# CORS(app)  # اجازه به همه دامنه‌ها
# CORS(app, origins=['http://localhost:3000'])  # فقط دامنه خاص

# ===== Middleware سفارشی (قبل از هر درخواست) =====
@app.before_request
def before_request():
    """لاگ کردن درخواست‌ها"""
    print(f"📨 {request.method} {request.path}")
    # می‌توانید برای اعتبارسنجی یا احراز هویت استفاده کنید

@app.after_request
def after_request(response):
    """بعد از هر درخواست (مثلاً برای CORS)"""
    # اضافه کردن هدرهای CORS
    response.headers.add('Access-Control-Allow-Origin', '*')
    response.headers.add('Access-Control-Allow-Headers', 'Content-Type')
    response.headers.add('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE')
    return response

# ===== مسیر نمونه =====
@app.route('/api/test')
def test():
    return jsonify({'message': 'CORS is enabled!'})

print("CORS and middleware are ready!")
print("  GET /api/test")

🔹 API با پایگاه داده (SQLite)

import sqlite3
from flask import Flask, request, jsonify

app = Flask(__name__)

# ===== تنظیم دیتابیس =====
def get_db_connection():
    conn = sqlite3.connect('database.db')
    conn.row_factory = sqlite3.Row
    return conn

def init_db():
    """ایجاد جدول کاربران"""
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT NOT NULL UNIQUE,
            age INTEGER
        )
    ''')
    conn.commit()
    conn.close()

# ===== مسیرها =====
@app.route('/api/db/users')
def get_users_db():
    conn = get_db_connection()
    users = conn.execute('SELECT * FROM users').fetchall()
    conn.close()
    return jsonify([dict(user) for user in users])

@app.route('/api/db/users', methods=['POST'])
def create_user_db():
    data = request.get_json()
    if not data or 'name' not in data or 'email' not in data:
        return jsonify({'error': 'Name and email are required'}), 400
    
    conn = get_db_connection()
    try:
        cursor = conn.execute(
            'INSERT INTO users (name, email, age) VALUES (?, ?, ?)',
            (data['name'], data['email'], data.get('age'))
        )
        conn.commit()
        user_id = cursor.lastrowid
        user = conn.execute('SELECT * FROM users WHERE id = ?', (user_id,)).fetchone()
        conn.close()
        return jsonify(dict(user)), 201
    except sqlite3.IntegrityError:
        conn.close()
        return jsonify({'error': 'Email already exists'}), 400

# ===== مقداردهی اولیه =====
# init_db()
print("Database API is ready!")
print("  GET  /api/db/users")
print("  POST /api/db/users")

🔹 JWT Authentication

# نصب (در محیط واقعی)
# pip install pyjwt
# pip install bcrypt

import jwt
import bcrypt
import datetime
from flask import Flask, request, jsonify
from functools import wraps

app = Flask(__name__)
SECRET_KEY = 'your-secret-key-here'  # در محیط واقعی از متغیر محیطی استفاده کنید

# ===== تولید توکن =====
def create_token(user_id, username):
    payload = {
        'user_id': user_id,
        'username': username,
        'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=1)
    }
    return jwt.encode(payload, SECRET_KEY, algorithm='HS256')

# ===== اعتبارسنجی توکن =====
def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('Authorization')
        
        if not token:
            return jsonify({'error': 'Token is missing'}), 401
        
        try:
            # حذف "Bearer " از ابتدای توکن
            token = token.split(' ')[1] if ' ' in token else token
            data = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
            request.user = data
        except jwt.ExpiredSignatureError:
            return jsonify({'error': 'Token has expired'}), 401
        except jwt.InvalidTokenError:
            return jsonify({'error': 'Invalid token'}), 401
        
        return f(*args, **kwargs)
    return decorated

# ===== مسیر ورود (دریافت توکن) =====
@app.route('/api/login', methods=['POST'])
def login():
    data = request.get_json()
    # اینجا باید کاربر را از دیتابیس بررسی کنید
    # برای مثال، یک کاربر نمونه
    if data and data.get('username') == 'admin' and data.get('password') == 'admin123':
        token = create_token(1, 'admin')
        return jsonify({
            'token': token,
            'message': 'Login successful'
        })
    return jsonify({'error': 'Invalid credentials'}), 401

# ===== مسیر محافظت‌شده =====
@app.route('/api/protected')
@token_required
def protected():
    return jsonify({
        'message': 'You have access to protected resource!',
        'user': request.user
    })

print("JWT Authentication is ready!")
print("  POST /api/login")
print("  GET  /api/protected (with Authorization header)")

🔹 مستندسازی با Swagger (فقط کپی)

# نصب (در محیط واقعی)
# pip install flasgger

from flask import Flask, jsonify
from flasgger import Swagger

app = Flask(__name__)
swagger = Swagger(app)

@app.route('/api/docs/hello')
def hello():
    """
    Endpoint سلام ساده
    ---
    responses:
      200:
        description: پیام خوش‌آمدگویی
        schema:
          type: object
          properties:
            message:
              type: string
              example: Hello, World!
    """
    return jsonify({'message': 'Hello, World!'})

print("Swagger documentation is ready!")
print("  GET /api/docs/hello")
print("  Visit /apidocs for Swagger UI")

🔹 مثال جامع: API مدیریت فروشگاه

from flask import Flask, request, jsonify
from datetime import datetime

app = Flask(__name__)

# ===== داده در حافظه =====
class ProductManager:
    def __init__(self):
        self.products = [
            {'id': 1, 'name': 'Laptop', 'price': 1200, 'stock': 10},
            {'id': 2, 'name': 'Phone', 'price': 800, 'stock': 25},
            {'id': 3, 'name': 'Tablet', 'price': 500, 'stock': 15}
        ]
        self.next_id = 4
        self.orders = []
        self.order_id = 1
    
    def get_all_products(self):
        return jsonify({
            'products': self.products,
            'count': len(self.products),
            'timestamp': datetime.now().isoformat()
        })
    
    def get_product(self, product_id):
        product = next((p for p in self.products if p['id'] == product_id), None)
        if not product:
            return jsonify({'error': 'Product not found'}), 404
        return jsonify(product)
    
    def create_product(self, data):
        if not data or 'name' not in data or 'price' not in data:
            return jsonify({'error': 'Name and price are required'}), 400
        
        product = {
            'id': self.next_id,
            'name': data['name'],
            'price': float(data['price']),
            'stock': int(data.get('stock', 0))
        }
        self.products.append(product)
        self.next_id += 1
        return jsonify(product), 201
    
    def update_product(self, product_id, data):
        product = next((p for p in self.products if p['id'] == product_id), None)
        if not product:
            return jsonify({'error': 'Product not found'}), 404
        
        if 'name' in data:
            product['name'] = data['name']
        if 'price' in data:
            product['price'] = float(data['price'])
        if 'stock' in data:
            product['stock'] = int(data['stock'])
        
        return jsonify(product)
    
    def delete_product(self, product_id):
        product = next((p for p in self.products if p['id'] == product_id), None)
        if not product:
            return jsonify({'error': 'Product not found'}), 404
        
        self.products = [p for p in self.products if p['id'] != product_id]
        return jsonify({'message': 'Product deleted successfully'})
    
    def create_order(self, data):
        if not data or 'product_id' not in data or 'quantity' not in data:
            return jsonify({'error': 'Product ID and quantity are required'}), 400
        
        product = next((p for p in self.products if p['id'] == data['product_id']), None)
        if not product:
            return jsonify({'error': 'Product not found'}), 404
        
        if product['stock'] < data['quantity']:
            return jsonify({'error': 'Insufficient stock'}), 400
        
        # کاهش موجودی
        product['stock'] -= data['quantity']
        
        # ایجاد سفارش
        order = {
            'id': self.order_id,
            'product_id': data['product_id'],
            'product_name': product['name'],
            'quantity': data['quantity'],
            'total': product['price'] * data['quantity'],
            'timestamp': datetime.now().isoformat()
        }
        self.orders.append(order)
        self.order_id += 1
        
        return jsonify(order), 201
    
    def get_all_orders(self):
        return jsonify({
            'orders': self.orders,
            'count': len(self.orders)
        })

# ===== ایجاد نمونه =====
pm = ProductManager()

# ===== مسیرها =====
@app.route('/api/store/products', methods=['GET'])
def get_all_products():
    return pm.get_all_products()

@app.route('/api/store/products/<int:product_id>', methods=['GET'])
def get_product(product_id):
    return pm.get_product(product_id)

@app.route('/api/store/products', methods=['POST'])
def create_product():
    return pm.create_product(request.get_json())

@app.route('/api/store/products/<int:product_id>', methods=['PUT'])
def update_product(product_id):
    return pm.update_product(product_id, request.get_json())

@app.route('/api/store/products/<int:product_id>', methods=['DELETE'])
def delete_product(product_id):
    return pm.delete_product(product_id)

@app.route('/api/store/orders', methods=['POST'])
def create_order():
    return pm.create_order(request.get_json())

@app.route('/api/store/orders', methods=['GET'])
def get_all_orders():
    return pm.get_all_orders()

print("=" * 50)
print("STORE API")
print("=" * 50)
print("Products:")
print("  GET    /api/store/products")
print("  GET    /api/store/products/<id>")
print("  POST   /api/store/products")
print("  PUT    /api/store/products/<id>")
print("  DELETE /api/store/products/<id>")
print("Orders:")
print("  GET    /api/store/orders")
print("  POST   /api/store/orders")

📌 نکات کلیدی

✅ Flask یک فریم‌ورک سبک و انعطاف‌پذیر برای ساخت APIها است .
✅ با `@app.route()` مسیرها را تعریف می‌کنید و متدهای HTTP را مشخص می‌کنید.
✅ `request.get_json()` برای دریافت داده JSON از بدنه درخواست استفاده می‌شود.
✅ `request.args` برای دریافت پارامترهای query string استفاده می‌شود.
✅ با `return jsonify()` پاسخ JSON را برمی‌گردانید.
✅ کدهای وضعیت HTTP را به عنوان آرگومان دوم به `return` اضافه کنید (مثلاً `201` یا `404`).
✅ `abort()` برای مدیریت خطاها و `@app.errorhandler()` برای سفارشی‌سازی خطاها استفاده می‌شود.
✅ Blueprint برای سازماندهی بهتر کد در پروژه‌های بزرگ استفاده می‌شود.
✅ flask-cors برای مدیریت CORS و JWT برای احراز هویت استفاده می‌شود.
✅ Flask برای پروژه‌های کوچک و متوسط و محصولات MVP بسیار مناسب است.

← قبلی: مبانی OpenCV → بعدی: فهرست پیشرفته ↑ بازگشت به فهرست پیشرفته