كيفية إنشاء واجهة برمجة تطبيقات لأسعار الصرف باستخدام FastAPI (httpx غير المتزامن، التخزين المؤقت، Pydantic)

إذا كنت تبني واجهة خلفية لتقنية مالية، أو صفحة دفع متعددة العملات، أو خدمة تسعير داخلية بلغة Python، فستحتاج عاجلاً أم آجلاً إلى نقطة نهاية خاصة بك لصرف العملات. إن بناء واجهة برمجة تطبيقات لأسعار الصرف باستخدام FastAPI يمنحك خدمة مصغّرة سريعة وغير متزامنة ومحدّدة الأنواع بالكامل، تغلّف مزوّد أسعار حيًّا، وتضيف تخزينًا مؤقتًا لتبقى ضمن الطبقة المجانية، وتعرض نقطة نهاية /convert نظيفة يمكن لخدماتك الأخرى استدعاؤها. يناسب FastAPI هذا السياق بشكل طبيعي: فهو مبني على ASGI من أجل إدخال/إخراج غير متزامن حقيقي، ويتحقّق من الطلبات والاستجابات باستخدام Pydantic، ويولّد وثائق OpenAPI تفاعلية مجانًا. يقطع هذا الدليل الطريق كاملاً — من أول طلب لك إلى المزوّد وصولاً إلى محوّل جاهز للإنتاج بعميل HTTP مشترك، وقراءات مخزّنة مؤقتًا، ومبالغ آمنة عبر Decimal، ومعالجة أخطاء محدّدة الأنواع.

في النهاية سيكون لديك خدمة FastAPI صغيرة ومكتفية ذاتيًا، مدعومة بـوثائق واجهة برمجة تطبيقات Finexly وتغطيتها لأكثر من 170 عملة. إن كنت قد قرأت بالفعل دليل واجهة برمجة تطبيقات العملات بلغة Python، فهذا هو المرافق على مستوى الخدمة الذي يبيّن كيف تتلاءم القطع في واجهة خلفية ASGI حقيقية. وبينما يعتمد دليل Django على الـORM وطبقة القوالب، يُبقي FastAPI كل شيء رشيقًا وموجّهًا نحو غير المتزامن.

لماذا يُعدّ FastAPI خيارًا رائعًا لخدمة عملات مصغّرة

يبدو تحويل العملات تافهًا — ضرب مبلغ في سعر — لكن القيام به بإتقان كخدمة يمسّ اهتمامات صُمّم FastAPI من أجلها:

إدخال/إخراج غير متزامن حقيقي. جلب الأسعار مقيّد بالشبكة. مع نقاط نهاية async def وعميل HTTP غير متزامن، يستطيع عامل واحد خدمة العديد من التحويلات المتزامنة بينما الطلبات جارية، بدلاً من حجب خيط لكل استدعاء.
عقود محدّدة الأنواع. تتحقّق نماذج Pydantic من معلمات الاستعلام الواردة وتصوغ JSON الصادر، فيُرفض amount مشوّه أو رمز عملة مجهول باستجابة 422 واضحة قبل أن يبلغ منطقك.
وثائق تفاعلية مجانية. تظهر كل نقطة نهاية تكتبها في واجهة Swagger على /docs، ما يجعل الخدمة موثّقة ذاتيًا لبقية فريقك.
إدارة lifespan. يمنحك FastAPI مكانًا نظيفًا لفتح مجمّع اتصالات HTTP واحد عند البدء وتصريفه عند الإيقاف — أمر حاسم للأداء، كما سترى أدناه.

المعمارية التي نبنيها هي وكيل رفيع مدرك للتخزين المؤقت: تستدعي خدمتك واجهة الأسعار، وتخزّن الاستجابة مؤقتًا لنافذة قصيرة، وتقدّم التحويلات لواجهتك الأمامية أو لخدمات مصغّرة أخرى. يبقي هذا النمط مفاتيح واجهتك على جانب الخادم، ويعزلك عن حدود معدّل المزوّد، ويتيح لك إضافة قواعد عمل (تقريب، هوامش، قوائم سماح) في مكان واحد.

المتطلبات المسبقة وإعداد المشروع

ستحتاج إلى Python 3.11 أو أحدث. أنشئ بيئة افتراضية وثبّت الاعتماديات:

python -m venv .venv
source .venv/bin/activate  # on Windows: .venv\Scripts\activate

pip install "fastapi[standard]" httpx pydantic-settings

يجلب fastapi[standard] خادم Uvicorn (خادم ASGI) وواجهة سطر الأوامر. أما httpx فهو عميل HTTP غير المتزامن لدينا، وpydantic-settings يتولّى الإعداد من متغيرات البيئة.

احصل على مفتاح واجهة مجاني من لوحة تحكم Finexly وصدّره كي لا يصل أبدًا إلى نظام التحكم بالإصدارات:

export FINEXLY_API_KEY="your_api_key_here"

أنشئ بنية المشروع:

fx-service/
├── app/
│   ├── __init__.py
│   ├── config.py
│   ├── client.py
│   ├── models.py
│   └── main.py
└── .env

الخطوة 1: الإعداد باستخدام pydantic-settings

أبقِ الإعداد في مكان واحد محدّد الأنواع. يقرأ pydantic-settings متغيرات البيئة وملف .env، ويفشل بصوت عالٍ إذا كانت قيمة مطلوبة مفقودة.

# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="FINEXLY_")

    api_key: str
    base_url: str = "https://api.finexly.com/v1"
    cache_ttl_seconds: int = 300
    request_timeout_seconds: float = 10.0


settings = Settings()  # raises at startup if FINEXLY_API_KEY is unset

بما أن كائن الإعدادات يُنشأ في وقت الاستيراد، فإن مفتاحًا مفقودًا يمنع الخدمة من الإقلاع بدلاً من الفشل عند أول طلب — وهو بالضبط ما تريده في خط أنابيب النشر.

الخطوة 2: عميل HTTP غير متزامن مشترك عبر lifespan

هذا هو أهم قرار يتعلق بالأداء في الخدمة بأكملها. لا تُنشئ httpx.AsyncClient جديدًا لكل طلب. يملك كل عميل مجمّع اتصالات؛ وإنشاء واحد لكل استدعاء يهدر اتصالات keep-alive ويفرض مصافحة TLS جديدة في كل مرة. بدلاً من ذلك، افتح عميلاً واحدًا عند بدء التطبيق وأعد استخدامه طوال عمر العملية.

الطريقة الحديثة لفعل ذلك هي مدير سياق lifespan في FastAPI. (أُهملت المُزخرِفات الأقدم @app.on_event("startup") و@app.on_event("shutdown") في FastAPI 0.93+ لصالح lifespan.) يضمن استخدام async with تصريف مجمّع الاتصالات وإغلاق المقابس عند الإيقاف، حتى لو قُوطع البدء.

# app/client.py
import httpx
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request

from .config import settings


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Runs once on startup: open a single pooled client
    async with httpx.AsyncClient(
        base_url=settings.base_url,
        headers={"Authorization": f"Bearer {settings.api_key}"},
        timeout=settings.request_timeout_seconds,
    ) as client:
        app.state.http = client
        yield
    # On shutdown, the async-with block closes the pool automatically


def get_client(request: Request) -> httpx.AsyncClient:
    # FastAPI dependency: hand endpoints the shared client
    return request.app.state.http

الآن تستعير كل نقطة نهاية العميل المجمّع ذاته عبر اعتمادية، ولا يحتاج أحد إلى التفكير في فتح الاتصالات أو إغلاقها.

الخطوة 3: نماذج طلب واستجابة محدّدة الأنواع

تمنحك نماذج Pydantic التحقّق واستجابات موثّقة ذاتيًا. لاحظ استخدام Decimal للمبالغ المالية: لا تستطيع أعداد الفاصلة العائمة الثنائية تمثيل الكسور العشرية بدقة، وأخطاء التقريب في المال خطأ برمجي حقيقي لا طرفة.

# app/models.py
from decimal import Decimal
from pydantic import BaseModel, Field


class ConversionResult(BaseModel):
    base: str = Field(..., examples=["USD"])
    quote: str = Field(..., examples=["EUR"])
    amount: Decimal = Field(..., examples=["100.00"])
    rate: Decimal = Field(..., examples=["0.92"])
    converted: Decimal = Field(..., examples=["92.00"])
    as_of: str = Field(..., description="Timestamp of the upstream rate")


class RatesResponse(BaseModel):
    base: str
    rates: dict[str, Decimal]
    as_of: str

سيعرض FastAPI هذه النماذج في مخطّط OpenAPI ويقسر/يتحقّق من القيم تلقائيًا، فيُرفض طلب بمبلغ لا معنى له قبل أن يعمل معالِجك.

الخطوة 4: جلب الأسعار بتخزين مؤقت TTL صغير

لا تتغيّر الأسعار من مِلّي ثانية لأخرى، وقرع واجهة المزوّد عند كل طلب بطيء وطريقة سريعة لاستنزاف حصتك. يخفّف تخزينٌ مؤقت صغير في الذاكرة بمدة حياة (TTL) هذا. لخدمة أحادية العملية، يكفي هذا التخزين المعتمد على قاموس؛ ولعدة عمّال، استبدله بـRedis لاحقًا دون تغيير مواضع الاستدعاء.

# app/main.py
import time
from decimal import Decimal, ROUND_HALF_UP

import httpx
from fastapi import Depends, FastAPI, HTTPException, Query

from .client import lifespan, get_client
from .config import settings
from .models import ConversionResult

app = FastAPI(title="FX Service", lifespan=lifespan)

# Simple in-memory cache: {base_currency: (expiry_timestamp, rates_dict)}
_cache: dict[str, tuple[float, dict]] = {}


async def fetch_rates(base: str, client: httpx.AsyncClient) -> dict:
    base = base.upper()
    now = time.monotonic()
    cached = _cache.get(base)
    if cached and cached[0] > now:
        return cached[1]  # cache hit — no network call

    try:
        resp = await client.get("/latest", params={"base": base})
        resp.raise_for_status()
    except httpx.HTTPStatusError as exc:
        code = exc.response.status_code
        if code == 429:
            raise HTTPException(503, "Upstream rate limit reached; try again shortly")
        if code in (401, 403):
            raise HTTPException(502, "Upstream authentication failed")
        raise HTTPException(502, "Upstream rates provider error")
    except httpx.RequestError:
        raise HTTPException(504, "Could not reach rates provider")

    rates = resp.json()["rates"]
    _cache[base] = (now + settings.cache_ttl_seconds, rates)
    return rates

لاحظ كيف تُترجَم إخفاقات المزوّد إلى رموز حالة HTTP ذات معنى لـمستدعيك. يصبح 429 من المزوّد استجابة 503 مع تلميح لإعادة المحاولة؛ ويصبح إخفاق المصادقة استجابة 502 كي لا تُسرّب أبدًا تفاصيل بيانات الاعتماد إلى الأسفل. يتعمّق دليلنا المرافق حول التخزين المؤقت ومعالجة الأخطاء في واجهات العملات في هذه الأنماط.

الخطوة 5: نقطتا النهاية /convert و/rates

الآن صارت نقاط النهاية نفسها ضئيلة، لأن كل العمل الشاق يعيش في fetch_rates. نتحقّق من معلمات الاستعلام عبر Query في FastAPI، ونجري حسابًا آمنًا عشريًا، ونعيد نماذج محدّدة الأنواع.

@app.get("/convert", response_model=ConversionResult)
async def convert(
    base: str = Query(..., min_length=3, max_length=3),
    quote: str = Query(..., min_length=3, max_length=3),
    amount: Decimal = Query(..., gt=0),
    client: httpx.AsyncClient = Depends(get_client),
):
    base, quote = base.upper(), quote.upper()
    rates = await fetch_rates(base, client)

    if quote not in rates:
        raise HTTPException(400, f"Unsupported currency: {quote}")

    rate = Decimal(str(rates[quote]))
    converted = (amount * rate).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)

    return ConversionResult(
        base=base, quote=quote, amount=amount,
        rate=rate, converted=converted, as_of="latest",
    )


@app.get("/health")
async def health():
    return {"status": "ok"}

شغّلها محليًا:

fastapi dev app/main.py

ثم افتح http://127.0.0.1:8000/docs لواجهة Swagger التفاعلية، أو استدعِ نقطة النهاية مباشرة:

curl "http://127.0.0.1:8000/convert?base=USD&quote=EUR&amount=100"

{
  "base": "USD",
  "quote": "EUR",
  "amount": "100",
  "rate": "0.92",
  "converted": "92.00",
  "as_of": "latest"
}

لديك الآن خدمة عملات مصغّرة تعمل. وإذا كنت تفضّل ألّا تشغّل حساب التحويل بنفسك مطلقًا، فتوفّر Finexly أيضًا محوّل عملات مُستضافًا ونقطة تحويل مباشرة يمكنك استدعاؤها.

الخطوة 6: ملاحظة حول الدقة العشرية والعملات بلا كسور

مصيدتان ماليتان توقعان بكل خدمة عملات تقريبًا:

لا تستخدم أبدًا الفاصلة العائمة الثنائية للمبالغ. 0.1 + 0.2 ليس 0.3 في حساب الفاصلة العائمة. حلّل الأسعار والمبالغ عبر Decimal(str(value)) وكمّم النتيجة النهائية، كما فعلنا أعلاه.
ليست كل عملة بمنزلتين عشريتين. يعرّف معيار ISO 4217 وحدات صغرى لكل عملة: JPY وKRW لهما صفر منزلة عشرية، بينما لـBHD وKWD ثلاث. إن ترميز .quantize(Decimal("0.01")) بشكل ثابت سيقرّب الين والدينار بشكل خاطئ بصمت. ينبغي لخدمة الإنتاج البحث عن الأُس الصحيح لكل عملة والتكميم وفقًا له.

إتقان هذا هو الفرق بين عرض تجريبي وشيء يمكنك وضعه أمام معاملات حقيقية.

الخطوة 7: اختبار الخدمة

بما أن عميل HTTP يُحقن كاعتمادية، فالاختبارات سهلة — تجاوز الاعتمادية بكائن وهمي ولن تلمس الشبكة أبدًا:

# test_convert.py
from fastapi.testclient import TestClient
from app.main import app, fetch_rates

async def fake_rates(base, client):
    return {"EUR": "0.92", "GBP": "0.79"}

def test_convert(monkeypatch):
    monkeypatch.setattr("app.main.fetch_rates", fake_rates)
    with TestClient(app) as c:
        r = c.get("/convert?base=USD&quote=EUR&amount=100")
        assert r.status_code == 200
        assert r.json()["converted"] == "92.00"

يشغّل TestClient أحداث lifespan نيابةً عنك، فيُعدّ العميل المشترك ويُفكّك تمامًا كما في الإنتاج.

نحو الإنتاج: التوسّع وحدود المعدّل

بعض الأمور للتخطيط لها قبل الإطلاق:

عمّال متعدّدون. تحت Uvicorn/Gunicorn بعدة عمّال، يكون تخزينك المؤقت في الذاكرة لكل عملية. انتقل إلى Redis لتخزين مشترك، كي يخدم سعرٌ جلبه عاملٌ واحد الجميع. يجعل المفصل fetch_rates هذا تغييرًا في دالة واحدة.
حصة المزوّد. التخزين المؤقت هو خط دفاعك الأول. بمدة TTL خمس دقائق، تكلّف عملة أساس واحدة 12 استدعاءً للمزوّد كحدٍّ أقصى في الساعة، بصرف النظر عن عدد التحويلات التي تقدّمها. راجع حدود خطتك في صفحة الأسعار واضبط TTL بما يلائم طبقتك.
الأسعار التاريخية. للفواتير أو التقارير أو مسارات التدقيق، ستريد نقطة نهاية /historical تستعلم عن أسعار تاريخ محدّد. تنطبق أنماط العميل والتخزين المؤقت ذاتها؛ فقط فهرِس التخزين بـ(base, date).
قابلية المراقبة. سجّل نسب الإصابة/الإخفاق للتخزين المؤقت وزمن استجابة المزوّد لضبط TTL ببيانات حقيقية.

إن كنت لا تزال تختار مزوّدًا، فصفحتنا لـمقارنة واجهات العملات تعرض التغطية والحدود والأسعار جنبًا إلى جنب.

الأسئلة الشائعة

لماذا نستخدم httpx بدلاً من requests في FastAPI؟

requests متزامن ويحجب حلقة الأحداث، ما يُبطل غرض إطار عمل غير متزامن. يوفّر httpx عميلاً غير متزامن بالكامل (httpx.AsyncClient) بمجمّع اتصالات ومهل زمنية ووسائط نقل بإعادة محاولة، فيندمج مباشرة في نقاط نهاية async def في FastAPI دون حجب الطلبات الأخرى.

هل أُنشئ عميل httpx لكل طلب أم مرة واحدة؟

مرة واحدة. أنشئ httpx.AsyncClient واحدًا في مدير سياق lifespan وأعد استخدامه عبر كل الطلبات. عميلٌ لكل طلب يهدر مجمّع الاتصالات ويفرض مصافحة TLS جديدة عند كل استدعاء، وهو أبطأ بشكل قابل للقياس تحت الحمل وقد يستنزف المقابس.

كيف أتجنّب بلوغ حد معدّل واجهة العملات؟

خزّن استجابات المزوّد مؤقتًا بمدة TTL قصيرة (خمس دقائق مثلاً). لأن الأسعار تتحرّك ببطء، يقلّص التخزين لكل عملة أساس آلاف تحويلات المستخدمين إلى حفنة استدعاءات للمزوّد في الساعة. لعمليات النشر متعدّدة العمّال، استخدم Redis كي يكون التخزين مشتركًا بين العمليات.

كيف أتعامل مع عملات بأعداد مختلفة من المنازل العشرية؟

استخدم نوع Decimal في Python، لا float أبدًا، وكمّم النتيجة إلى العدد الصحيح من الوحدات الصغرى لكل عملة وفق ISO 4217. تستخدم JPY وKRW صفر منزلة، ويستخدم معظمها منزلتين، وتستخدم BHD/KWD ثلاثًا. ابحث عن الأُس لكل عملة بدلاً من ترميز منزلتين بشكل ثابت.

هل يمكنني استخدام هذا النمط مع إطار عمل متزامن مثل Flask أو Django؟

تنتقل أنماط التخزين المؤقت وDecimal مباشرة، لكن أجزاء العميل المشترك ونقطة النهاية غير المتزامنة خاصة بـFastAPI. لـDjango، انظر دليل واجهة برمجة تطبيقات العملات المخصّص لـDjango، الذي يستخدم واجهة التخزين المؤقت والـORM الخاصين بالإطار.

ابدأ البناء

لديك الآن خدمة عملات مصغّرة كاملة وغير متزامنة ومدركة للتخزين المؤقت في FastAPI — محدّدة الأنواع بـPydantic، وآمنة بـDecimal، وصامدة أمام إخفاقات المزوّد. إنها صغيرة بما يكفي لقراءتها في جلسة واحدة، ومنظّمة بما يكفي لتنمو إلى خدمة إنتاج حقيقية.

هل أنت جاهز لدمج أسعار الصرف الفورية في مشروعك؟ احصل على مفتاح Finexly المجاني — دون بطاقة ائتمان. ابدأ بطبقة مجانية سخية تغطّي أكثر من 170 عملة، ووسّع مع نمو حركة مرورك. تفضّل الاستكشاف أولاً؟ تصفّح نظرة عامة على واجهة العملات المجانية أو اغمر في وثائق الواجهة.