دریافت امن نرخ ارز با API سرویکس در Python

این راهنما یک اتصال سمت بک اند با Python برای API احرازشده داده بازار Servix می سازد. نمونه از قرارداد supported-assets استفاده می کند، زیرا در یک درخواست کد نماد، آخرین مقدار احرازشده، businessTime، پوشش تامین کننده، دسترس پذیری و پرچم stale محاسبه شده در سرور را در اختیار برنامه قرار می دهد.

کد را فقط در سرور، worker یا job زمان بندی شده اجرا کنید. API Key سرویکس و پاسخ احرازشده را در کد مرورگر، HTML عمومی، analytics، گزارش خطا یا مخزن عمومی قرار ندهید.

پیش نیازها و نصب

• Python 3.11 یا جدیدتر؛
• حساب فعال Servix و API Key ساخته شده در Profile > API Keys؛
• نصب کتابخانه requests در virtual environment؛
• secret store سمت سرور یا متغیر محیطی برای نگهداری کلید.

python -m venv .venv
source .venv/bin/activate
python -m pip install "requests>=2.32,<3"

export SERVIX_API_KEY="<your-servix-api-key>"
export SERVIX_ASSET_CODE="USD_RLS"
python servix_currency_api.py

API Key فقط در هدر X-API-Key ارسال می شود و هرگز به URL یا لاگ اضافه نمی شود. برای توسعه و تولید کلیدهای جدا بسازید و در صورت احتمال افشا، کلید را فوری لغو کنید.

نمونه کامل و قابل اجرا

نسخه بررسی شده در مسیر docs/examples/servix_currency_api.py قرار دارد. تست های قرارداد با fixture مصنوعی کنار همان فایل هستند و هیچ درخواستی به API تولید ارسال نمی کنند.

"""Fetch and validate one Servix market quote from a trusted Python backend."""

from __future__ import annotations

import logging
import os
import time
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from datetime import datetime
from typing import Any, Callable

import requests

DEFAULT_BASE_URL = "https://servix.cc"
SUPPORTED_ASSETS_PATH = "/api/v1/assets/supported"
CONNECT_TIMEOUT_SECONDS = 3.05
READ_TIMEOUT_SECONDS = 10
MAXIMUM_ATTEMPTS = 3
RETRYABLE_STATUS_CODES = frozenset({500, 502, 503, 504})

LOGGER = logging.getLogger("servix.currency_api")


class ServixError(RuntimeError):
    """Base exception for safe, actionable Servix client failures."""


class ServixConfigurationError(ServixError):
    """Raised when required local configuration is missing."""


class ServixNetworkError(ServixError):
    """Raised after bounded network retries are exhausted."""


class ServixHttpError(ServixError):
    """Raised for a non-retryable Servix HTTP response."""


class ServixRateLimitError(ServixHttpError):
    """Raised when the account's daily request quota is exhausted."""


class ServixSchemaError(ServixError):
    """Raised when the authenticated response does not match the expected schema."""


class ServixUnavailableError(ServixError):
    """Raised when the requested asset has no usable latest observation."""


class ServixStaleDataError(ServixError):
    """Raised when Servix marks the latest observation as stale."""


@dataclass(frozen=True)
class AssetQuote:
    code: str
    label: str
    value: Decimal
    business_time: datetime
    unit: str


def require_api_key(environment: dict[str, str] | None = None) -> str:
    source = os.environ if environment is None else environment
    api_key = source.get("SERVIX_API_KEY", "").strip()
    if not api_key:
        raise ServixConfigurationError(
            "SERVIX_API_KEY is required; keep it in a server-side secret store."
        )
    return api_key


def fetch_asset_quote(
    asset_code: str = "USD_RLS",
    *,
    session: requests.Session | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    sleep: Callable[[float], None] = time.sleep,
) -> AssetQuote:
    normalized_code = asset_code.strip().upper()
    if not normalized_code:
        raise ServixConfigurationError("An asset code is required.")

    resolved_api_key = require_api_key() if api_key is None else api_key.strip()
    if not resolved_api_key:
        raise ServixConfigurationError("A non-empty Servix API key is required.")

    resolved_base_url = (
        base_url or os.environ.get("SERVIX_API_BASE_URL") or DEFAULT_BASE_URL
    ).rstrip("/")
    client = session or requests.Session()
    payload = _request_json(
        client,
        f"{resolved_base_url}{SUPPORTED_ASSETS_PATH}",
        resolved_api_key,
        sleep,
    )
    return _validate_asset(payload, normalized_code)


def _request_json(
    session: requests.Session,
    url: str,
    api_key: str,
    sleep: Callable[[float], None],
) -> Any:
    for attempt in range(1, MAXIMUM_ATTEMPTS + 1):
        try:
            response = session.get(
                url,
                headers={
                    "Accept": "application/json",
                    "X-API-Key": api_key,
                },
                timeout=(CONNECT_TIMEOUT_SECONDS, READ_TIMEOUT_SECONDS),
            )
        except (requests.ConnectionError, requests.Timeout) as exception:
            if attempt == MAXIMUM_ATTEMPTS:
                raise ServixNetworkError(
                    "Servix network request failed after bounded retries."
                ) from exception
            sleep(_retry_delay(attempt))
            continue

        request_id = response.headers.get("x-request-id", "unavailable")
        if response.status_code == 429:
            raise ServixRateLimitError(
                f"Servix daily quota is exhausted; requestId={request_id}."
            )
        if 400 <= response.status_code < 500:
            raise ServixHttpError(
                f"Servix rejected the request with HTTP {response.status_code}; "
                f"requestId={request_id}."
            )
        if response.status_code in RETRYABLE_STATUS_CODES:
            if attempt == MAXIMUM_ATTEMPTS:
                raise ServixHttpError(
                    f"Servix remained unavailable with HTTP {response.status_code}; "
                    f"requestId={request_id}."
                )
            sleep(_retry_delay(attempt))
            continue
        if not 200 <= response.status_code < 300:
            raise ServixHttpError(
                f"Servix returned unexpected HTTP {response.status_code}; "
                f"requestId={request_id}."
            )

        try:
            return response.json(parse_float=Decimal)
        except ValueError as exception:
            raise ServixSchemaError(
                f"Servix returned malformed JSON; requestId={request_id}."
            ) from exception

    raise ServixNetworkError("Servix request ended without a response.")


def _validate_asset(payload: Any, asset_code: str) -> AssetQuote:
    if not isinstance(payload, dict) or not isinstance(payload.get("assets"), list):
        raise ServixSchemaError("Servix response must contain an assets array.")

    asset = next(
        (
            candidate
            for candidate in payload["assets"]
            if isinstance(candidate, dict) and candidate.get("code") == asset_code
        ),
        None,
    )
    if asset is None:
        raise ServixUnavailableError(f"Servix does not list asset {asset_code}.")

    for field in ("providerSupported", "latestValueAvailable", "stale"):
        if not isinstance(asset.get(field), bool):
            raise ServixSchemaError(f"Servix field {field} must be boolean.")

    availability_status = asset.get("availabilityStatus")
    if availability_status not in {"FRESH", "STALE", "MISSING", "UNKNOWN"}:
        raise ServixSchemaError(
            "Servix availabilityStatus must be FRESH, STALE, MISSING, or UNKNOWN."
        )

    if not asset["providerSupported"] or not asset["latestValueAvailable"]:
        raise ServixUnavailableError(
            f"Servix has no usable latest observation for {asset_code}."
        )

    business_time = _parse_business_time(asset.get("businessTime"))
    if asset["stale"] != (availability_status == "STALE"):
        raise ServixSchemaError(
            "Servix stale and availabilityStatus fields are inconsistent."
        )
    if asset["stale"]:
        raise ServixStaleDataError(
            f"Servix marks {asset_code} as stale at {business_time.isoformat()}."
        )
    if availability_status != "FRESH":
        raise ServixUnavailableError(
            f"Servix marks {asset_code} as unavailable."
        )

    value = asset.get("latestValue")
    if isinstance(value, bool) or not isinstance(value, (int, float, Decimal)):
        raise ServixSchemaError("Servix latestValue must be numeric.")
    try:
        numeric_value = value if isinstance(value, Decimal) else Decimal(str(value))
    except InvalidOperation as exception:
        raise ServixSchemaError("Servix latestValue must be numeric.") from exception
    if not numeric_value.is_finite():
        raise ServixSchemaError("Servix latestValue must be finite.")

    label = asset.get("labelEn")
    if not isinstance(label, str) or not label.strip():
        label = asset_code

    return AssetQuote(
        code=asset_code,
        label=label,
        value=numeric_value,
        business_time=business_time,
        unit=_unit_for(asset_code),
    )


def _parse_business_time(value: Any) -> datetime:
    if not isinstance(value, str) or not value.strip():
        raise ServixSchemaError("Servix businessTime must be an ISO-8601 timestamp.")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exception:
        raise ServixSchemaError(
            "Servix businessTime must be an ISO-8601 timestamp."
        ) from exception
    if parsed.tzinfo is None:
        raise ServixSchemaError("Servix businessTime must include a timezone.")
    return parsed


def _unit_for(asset_code: str) -> str:
    if asset_code.endswith("_RLS"):
        return "rial"
    if asset_code.endswith("_USD"):
        return "USD"
    return "contract-defined unit"


def _retry_delay(attempt: int) -> float:
    return 0.5 * (2 ** (attempt - 1))


def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    quote = fetch_asset_quote(os.environ.get("SERVIX_ASSET_CODE", "USD_RLS"))
    LOGGER.info(
        "Validated Servix quote for %s at %s.",
        quote.code,
        quote.business_time.isoformat(),
    )
    print(f"{quote.label}: {quote.value:g} {quote.unit}")


if __name__ == "__main__":
    main()

قرارداد پاسخ احرازشده

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

{
  "code": "USD_RLS",
  "providerSupported": true,
  "latestValueAvailable": true,
  "latestValue": "<protected-market-value>",
  "businessTime": "<source-timestamp>",
  "availabilityStatus": "FRESH",
  "stale": false
}

• businessTime زمان منبع قیمت است، نه زمان پایان درخواست HTTP برنامه شما.
• providerSupported=false یعنی تامین کننده های پیکربندی شده در حال حاضر آن نماد را پوشش نمی دهند.
• latestValueAvailable=false یعنی مقدار قابل استفاده ای وجود ندارد؛ صفر یا مقدار ساختگی جایگزین نکنید.
• stale=true یعنی Servix داده را قدیمی تر از آستانه تازگی می داند. نمونه به جای تازه فرض کردن داده، اجرای عادی را متوقف می کند.

رفتار timeout، خطای HTTP، شبکه و schema

پیام های خطای نمونه فقط دسته امن وضعیت و request ID سرور را دارند و شامل API Key، query، بدنه پاسخ یا مقدار بازار نیستند. برای جزئیات کامل خطاها و محدودیت درخواست را بخوانید.

ریال، تومان، تازگی و cache سهمیه محور

نمادهای پایان یافته به _RLS بر حسب ریال ایران هستند. اگر محصول شما تومان نمایش می دهد، مقدار عددی احرازشده را فقط در مرز نمایش برنامه خود بر ۱۰ تقسیم کنید و واحد را روشن بنویسید. نمادهای پایان یافته به _USD دلاری هستند. واحد را از ترجمه برچسب حدس نزنید و قرارداد و روش شناسی Servix را مبنا قرار دهید.

پاسخ موفق احرازشده را در بک اند با کلید کد نماد و همراه businessTime cache کنید. TTL باید کوتاه تر از پنجره تازگی قابل قبول محصول باشد، cache میان کاربران به اشتراک گذاشته شود و برای هر مرورگر polling جدا انجام نشود. پاسخ stale یا unavailable را به صفر تازه نما تبدیل نکنید و مقدار قدیمی را بدون هشدار بازپخش نکنید.

مسیرهای بعدی

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

ساخت حساب و فعال سازی حساب تست · مقایسه پلن ها و سهمیه تولید