اعمل API من ملف CSV بـ FastAPI وDocker في 20 دقيقة

مستوى القارئ: متوسط

هتطلع من المقال بـ API شغال يحوّل ملف CSV عادي إلى JSON endpoint قابل للبحث، بدل ما تفتح Excel أو تبعت الملف يدويًا كل مرة.

المشكلة باختصار

لو عندك ملف products.csv بيتحدث مرة أو مرتين في اليوم، والطريقة الحالية إن حد يفتحه ويدور داخله، فأنت بتضيع وقت في شغل مكرر. السيناريو الواقعي: فريق مبيعات عنده 8000 منتج، وكل استعلام سعر بياخد من 3 إلى 15 دقيقة حسب حجم الملف. بعد API محلي بسيط، نفس الاستعلام ممكن يرد في أقل من 3 ثواني على جهاز عادي.

الطريقة الشائعة الغلط هنا إنك تبني قاعدة بيانات كاملة من أول يوم. ده ممكن يكون مبالغ فيه لو البيانات ثابتة وصغيرة. أفضل طريقة في الحالة دي: ابدأ بـ FastAPI يقرأ CSV ويطلع endpoint، وبعدها قرر هل محتاج PostgreSQL فعلاً ولا لا.

الفكرة بمثال بسيط

ركز في المثال ده. عندك ملف منتجات فيه الاسم والسعر والتصنيف. بدل ما تسأل شخص يفتح الملف، هتسأل endpoint زي /search?q=keyboard. الـ API يقرأ الصفوف، يفلتر النتائج، ويرجع JSON ينفع يتعرض في لوحة داخلية أو يتربط مع n8n أو سكربت آخر.

الافتراض إن الملف أقل من 50 ألف صف، وحجمه أقل من 20MB، والتحديث مش لحظي. لو عندك ملايين الصفوف أو تحديثات كل ثانية، هذا التصميم مش مناسب كحل نهائي.

1. جهز المشروع والملف

اعمل فولدر جديد، ثم ضع ملف products.csv بهذا الشكل:

id,name,category,price
1,Mechanical Keyboard,accessories,79.99
2,USB-C Hub,accessories,34.50
3,Standing Desk,office,299.00

بعدها أنشئ ملف requirements.txt:

fastapi==0.115.6
uvicorn[standard]==0.34.0

FastAPI مناسب هنا لأنه يولد توثيقًا تفاعليًا تلقائيًا على /docs، وده يقلل وقت شرح الـ endpoint لأي شخص في الفريق.

2. اكتب FastAPI يقرأ CSV

اكتب ملف main.py. المثال متعمد يكون قصير وواضح، لكنه عملي:

from csv import DictReader
from pathlib import Path
from fastapi import FastAPI, Query

app = FastAPI(title="CSV Products API")
DATA_FILE = Path("products.csv")

def load_products():
    with DATA_FILE.open(newline="", encoding="utf-8") as file:
        return list(DictReader(file))

PRODUCTS = load_products()

@app.get("/health")
def health():
    return {"status": "ok", "rows": len(PRODUCTS)}

@app.get("/search")
def search(q: str = Query(min_length=2), limit: int = 20):
    q = q.lower().strip()
    results = [
        item for item in PRODUCTS
        if q in item["name"].lower() or q in item["category"].lower()
    ]
    return {"count": len(results), "items": results[:limit]}

الـ DictReader يحول كل صف في CSV إلى dictionary. ده يخلي الكود أوضح من التعامل مع indexes رقمية. التكلفة: تحميل الملف كله في الذاكرة عند التشغيل. المكسب: زمن استجابة سريع وبساطة عالية.

3. شغل وجرب محليًا

ثبت الحزم وشغل السيرفر:

python -m venv .venv
. .venv/bin/activate
pip install -r requirements.txt
uvicorn main:app --reload --host 127.0.0.1 --port 8000

على Windows PowerShell استخدم:

.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt
uvicorn main:app --reload --host 127.0.0.1 --port 8000

اختبر النتيجة:

curl "http://127.0.0.1:8000/health"
curl "http://127.0.0.1:8000/search?q=keyboard"

لو فتحت http://127.0.0.1:8000/docs هتشوف توثيق Swagger تلقائي. ده مفيد جدًا لما تسلم الأداة لشخص غير تقني نسبيًا.

4. لفه داخل Docker

لو عايزها تدعم التشغيل على أي جهاز بدون اختلاف إصدارات Python، استخدم Dockerfile بسيط:

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY main.py products.csv ./
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

ثم شغل:

docker build -t csv-products-api .
docker run --rm -p 8000:8000 csv-products-api

الـ trade-off هنا واضح. Docker يزود حجم وتشغيل أول مرة، لكنه يقلل مشاكل “شغال عندي ومش شغال عندك”. لو المشروع داخلي لفريق صغير، المكسب غالبًا يستاهل.

القياس قبل وبعد

في تجربة داخلية بسيطة على ملف 8000 صف، البحث اليدوي عن منتج داخل spreadsheet كان بياخد بين 5 و15 دقيقة مع نسخ النتيجة. الـ endpoint المحلي رجع نفس النتيجة في 2 إلى 3 ثواني. الرقم هنا ليس benchmark عالمي، لكنه كافي لاتخاذ قرار: الأتمتة الصغيرة تستاهل لما العملية تتكرر يوميًا.

متى لا تستخدم هذه الطريقة

• لا تستخدمها لو البيانات تتغير كل ثانية. استخدم قاعدة بيانات.
• لا تستخدمها لو CSV أكبر من الذاكرة المتاحة. وقتها اقرأ على دفعات أو انتقل لـ SQLite/PostgreSQL.
• لا تعرضها على الإنترنت بدون authentication وrate limiting.
• لا تعتمد عليها كـ source of truth لو أكثر من شخص يعدل الملف في نفس الوقت.

مصادر اعتمدت عليها

• FastAPI First Steps لتشغيل تطبيق بسيط وتوثيق /docs.
• Python csv module لاستخدام DictReader مع ملفات CSV.
• Uvicorn settings لإعدادات host وport.
• Dockerfile reference لفهم FROM وCOPY وCMD.

الخطوة التالية

الخطوة التالية: خذ CSV حقيقي عندك، نفذ /health و/search، ثم قس زمن الوصول قبل وبعد. لو الملف أكبر من 20MB أو البحث بقى أبطأ من ثانية، انقل نفس الفكرة إلى SQLite بدل ما تكبر الكود عشوائيًا.