مرکز ابزارها، افزونه ها و مثال های اتصال

در این صفحه بسته های آماده برای اتصال به سرویس پرداخت، افزونه های کمکی و مثال های اجرایی را مثل یک ریپازیتوری مرور می کنید: فایل ها را می بینید، README را می خوانید و کل بسته را با یک دانلود می گیرید.

CONTROLLED RELEASE

انتشار کنترل شده از پنل رسمی

هر بسته از مسیر مدیریتی رسمی پی اکسا منتشر می شود و قبل از دانلود، اطلاعات کلیدی آن به صورت شفاف نمایش داده می شود.

PRE-DOWNLOAD VISIBILITY

شفافیت محتوا پیش از دریافت

پیش نمایش README، مرور ساختار فایل و نمایش متادیتای بسته کمک می کند قبل از دانلود، ارزیابی دقیق تری از محتوا داشته باشید.

TRUST PROFILE

نمای رسمی تر برای انتخاب ابزار

صفحه بر پایه شفافیت انتشار، نمایش منبع README، تاریخ بروزرسانی و آمار دانلود چیده شده تا حس ارزیابی حرفه ای تری منتقل کند.

مرجع حقوقی و هویتی کنار دسترس

ساختار صفحه طوری چیده شده که اطلاعات انتشار، محتوای فایل ها و جزئیات هر بسته بدون شلوغی اضافه در دسترس باشند.

مثال repository / pypi-sdk

مثال استفاده از كتابخانه payexa

استفاده از كتابخانه payexa از طريق نصب pip

دانلود بسته

6 فایل wallet-charge-demo.zip README: README.md

دانلود ثبت شده

آخرین بروزرسانی

2026-09-25

حجم بسته

7.5 KB

منتشرکننده

مدیر ارشد سامانه

وضعیت نمایش

عمومی و فعال

مرورگر فایل

پوشه جاری را انتخاب کنید یا برای فایل های متنی پیش نمایش بگیرید.

ریشه

نام	نوع	اندازه
templates	پوشه	-
.env.example	دانلودی	0.2 KB
app.py	قابل پیش نمایش	10.2 KB
README.md	قابل پیش نمایش	1.7 KB
requirements.txt	قابل پیش نمایش	0.0 KB

پیش نمایش فایل

app.py

from __future__ import annotations

import os
import sqlite3
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation

from dotenv import load_dotenv
from flask import Flask, flash, redirect, render_template, request, url_for
from payexa import PayexaAPIError, PayexaClient, PayexaTransportError, PayexaValidationError


def _read_env(name: str, default: str) -> str:
    value = str(os.getenv(name, "") or "").strip()
    return value or default


def _read_optional_int(name: str) -> int | None:
    value = str(os.getenv(name, "") or "").strip()
    if not value:
        return None
    return int(value)


def _resolve_database_path(database_path: str) -> str:
    normalized = str(database_path or "").strip()
    if not normalized:
        return os.path.join(app.instance_path, "wallet.sqlite3")
    if os.path.isabs(normalized):
        return normalized
    return os.path.abspath(os.path.join(app.instance_path, normalized))


load_dotenv()

app = Flask(__name__)
app.config["SECRET_KEY"] = os.getenv("FLASK_SECRET_KEY", "dev-secret-key")
app.config["PAYEXA_BASE_URL"] = _read_env("PAYEXA_BASE_URL", "https://pay.pexn.ir")
app.config["PAYEXA_API_KEY"] = _read_env("PAYEXA_API_KEY", "YOUR_API_KEY")
app.config["PAYEXA_MERCHANT_ID"] = _read_optional_int("PAYEXA_MERCHANT_ID")
app.config["APP_BASE_URL"] = _read_env("APP_BASE_URL", "http://127.0.0.1:5000")
app.config["WALLET_CURRENCY_LABEL"] = _read_env("WALLET_CURRENCY_LABEL", "تومان")
app.config["DATABASE_PATH"] = _resolve_database_path(
    _read_env("DATABASE_PATH", os.path.join(app.instance_path, "wallet.sqlite3"))
)

os.makedirs(app.instance_path, exist_ok=True)


@app.get("/")
def index() -> str:
    return render_template(
        "index.html",
        balance=get_wallet_balance(app.config["DATABASE_PATH"]),
        currency_label=app.config["WALLET_CURRENCY_LABEL"],
        payexa_base_url=app.config["PAYEXA_BASE_URL"],
    )


@app.post("/charge")
def charge_wallet():
    amount_raw = str(request.form.get("amount") or "").strip()

    try:
        amount = int(Decimal(amount_raw))
        if amount <= 0:
            raise ValueError
    except (InvalidOperation, ValueError):
        flash("مبلغ باید یک عدد صحیح و بزرگتر از صفر باشد.", "error")
        return redirect(url_for("index"))

    try:
        with build_client() as client:
            payment = client.request_payment(
                amount=amount,
                callback_url=build_callback_url(app.config["APP_BASE_URL"]),
                merchant_id=app.config["PAYEXA_MERCHANT_ID"],
                meta={"source": "test-pypi-flask-demo", "amount": amount},
            )
    except (PayexaValidationError, PayexaTransportError, PayexaAPIError) as exc:
        flash(f"ایجاد درخواست پرداخت ناموفق بود: {exc}", "error")
        return redirect(url_for("index"))

    save_payment(
        database_path=app.config["DATABASE_PATH"],
        authority=payment.authority,
        order_id=payment.order_id,
        amount=amount,
        amount_unique_token=payment.amount_unique_token,
        start_pay_url=payment.start_pay_url,
    )
    return redirect(payment.start_pay_url)


@app.get("/callback")
def callback() -> str:
    try:
        with build_client() as client:
            callback_data = client.parse_callback(request.args)
    except PayexaValidationError as exc:
        return render_template("result.html", ok=False, title="Callback نامعتبر", message=str(exc))

    if not callback_data.has_authority:
        return render_template(
            "result.html",
            ok=False,
            title="Authority پیدا نشد",
            message="پارامتر authority در callback ارسال نشده است.",
        )

    payment = get_payment_by_authority(app.config["DATABASE_PATH"], callback_data.authority)
    if payment is None:
        return render_template(
            "result.html",
            ok=False,
            title="تراکنش پیدا نشد",
            message="برای این authority هیچ رکوردی در دیتابیس این پروژه وجود ندارد.",
        )

    if not callback_data.is_ok:
        update_payment_status(app.config["DATABASE_PATH"], callback_data.authority, callback_data.normalized_status or "FAILED")
        return render_template(
            "result.html",
            ok=False,
            title="پرداخت ناموفق",
            message=f"وضعیت callback موفق نبود: {callback_data.normalized_status or 'UNKNOWN'}",
            payment=payment,
        )

    try:
        with build_client() as client:
            verification = client.verify_payment(
                order_id=payment["order_id"],
                token=payment["amount_unique_token"],
            )
    except (PayexaValidationError, PayexaTransportError, PayexaAPIError) as exc:
        return render_template(
            "result.html",
            ok=False,
            title="خطا در verify",
            message=f"verify انجام نشد: {exc}",
            payment=payment,
        )

    if verification.is_success:
        if payment["status"] != "verified":
            update_payment_status(app.config["DATABASE_PATH"], callback_data.authority, "verified")
            add_wallet_balance(app.config["DATABASE_PATH"], payment["amount"])
        payment = get_payment_by_authority(app.config["DATABASE_PATH"], callback_data.authority)
        return render_template(
            "result.html",
            ok=True,
            title="شارژ کیف پول موفق",
            message="پرداخت verify شد و موجودی کیف پول افزایش یافت.",
            payment=payment,
            verification_status=verification.normalized_status,
            balance=get_wallet_balance(app.config["DATABASE_PATH"]),
            currency_label=app.config["WALLET_CURRENCY_LABEL"],
        )

    update_payment_status(app.config["DATABASE_PATH"], callback_data.authority, verification.normalized_status or "FAILED")
    return render_template(
        "result.html",
        ok=False,
        title="verify ناموفق",
        message=f"نتیجه verify موفق نبود: {verification.normalized_status or 'UNKNOWN'}",
        payment=payment,
    )


def build_client() -> PayexaClient:
    return PayexaClient(
        base_url=app.config["PAYEXA_BASE_URL"],
        api_key=app.config["PAYEXA_API_KEY"],
        default_merchant_id=app.config["PAYEXA_MERCHANT_ID"],
    )


def build_callback_url(app_base_url: str) -> str:
    return f"{app_base_url.rstrip('/')}/callback"


def init_db(database_path: str) -> None:
    with connect_db(database_path) as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS wallet (
                id INTEGER PRIMARY KEY CHECK (id = 1),
                balance INTEGER NOT NULL DEFAULT 0
            )
            """
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS payments (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                authority TEXT NOT NULL UNIQUE,
                order_id TEXT NOT NULL,
                amount INTEGER NOT NULL,
                amount_unique_token TEXT NOT NULL,
                start_pay_url TEXT NOT NULL,
                status TEXT NOT NULL DEFAULT 'pending',
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )
            """
        )
        connection.execute("INSERT OR IGNORE INTO wallet (id, balance) VALUES (1, 0)")


def save_payment(
    database_path: str,
    *,
    authority: str,
    order_id: str,
    amount: int,
    amount_unique_token: str,
    start_pay_url: str,
) -> None:
    now = utc_now()
    with connect_db(database_path) as connection:
        connection.execute(
            """
            INSERT INTO payments (
                authority,
                order_id,
                amount,
                amount_unique_token,
                start_pay_url,
                status,
                created_at,
                updated_at
            ) VALUES (?, ?, ?, ?, ?, 'pending', ?, ?)
            ON CONFLICT(authority) DO UPDATE SET
                order_id = excluded.order_id,
                amount = excluded.amount,
                amount_unique_token = excluded.amount_unique_token,
                start_pay_url = excluded.start_pay_url,
                status = 'pending',
                updated_at = excluded.updated_at
            """,
            (authority, order_id, amount, amount_unique_token, start_pay_url, now, now),
        )


def get_payment_by_authority(database_path: str, authority: str) -> dict[str, object] | None:
    with connect_db(database_path) as connection:
        row = connection.execute(
            """
            SELECT authority, order_id, amount, amount_unique_token, start_pay_url, status, created_at, updated_at
            FROM payments WHERE authority = ?
            """,
            (authority,),
        ).fetchone()
    return dict(row) if row is not None else None


def get_wallet_balance(database_path: str) -> int:
    with connect_db(database_path) as connection:
        row = connection.execute("SELECT balance FROM wallet WHERE id = 1").fetchone()
    return int(row[0] if row is not None else 0)


def add_wallet_balance(database_path: str, amount: int) -> None:
    with connect_db(database_path) as connection:
        connection.execute("UPDATE wallet SET balance = balance + ? WHERE id = 1", (amount,))


def update_payment_status(database_path: str, authority: str, status: str) -> None:
    normalized_status = str(status or "failed").strip().casefold() or "failed"
    with connect_db(database_path) as connection:
        connection.execute(
            "UPDATE payments SET status = ?, updated_at = ? WHERE authority = ?",
            (normalized_status, utc_now(), authority),
        )


def connect_db(database_path: str) -> sqlite3.Connection:
    parent_dir = os.path.dirname(os.path.abspath(database_path))
    if parent_dir:
        os.makedirs(parent_dir, exist_ok=True)
    connection = sqlite3.connect(database_path)
    connection.row_factory = sqlite3.Row
    return connection


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


init_db(app.config["DATABASE_PATH"])


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

README / توضیحات

منبع: README.md

Flask Wallet Charge Demo

این پروژه یک دمو ساده Flask برای شارژ کیف پول با استفاده از پکیج payexa از PyPI است.

پیش نیاز

Python 3.10+
دسترسی شبکه به PAYEXA

نصب

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt

تنظیم env

cp .env.example .env

مقادیر مهم در .env:

PAYEXA_BASE_URL=https://pay.pexn.ir
PAYEXA_API_KEY=YOUR_API_KEY
PAYEXA_MERCHANT_ID=
APP_BASE_URL=http://127.0.0.1:5000

اگر DATABASE_PATH خالی باشد، پروژه از instance/wallet.sqlite3 استفاده می کند.

اجرا

source .venv/bin/activate
python app.py

بعد از اجرا اپ روی آدرس زیر در دسترس است:

http://127.0.0.1:5000

روند کار

مبلغ شارژ را در صفحه اصلی وارد می کنید.
اپ با PayexaClient درخواست پرداخت می سازد.
کاربر به درگاه منتقل می شود.
بعد از callback، اپ parse_callback و سپس verify_payment را اجرا می کند.
اگر verify موفق باشد، مبلغ به کیف پول SQLite اضافه می شود.

تست سریع

.venv/bin/python -c &quot;from payexa import PayexaClient; print(PayexaClient)&quot;
.venv/bin/python -c &quot;from app import app; client = app.test_client(); print(client.get(&#x27;/&#x27;).status_code)&quot;

نکته

برای callback واقعی، مقدار APP_BASE_URL باید از بیرون قابل دسترسی باشد. اگر روی localhost اجرا می کنید، فقط بخش local flow را راحت تست می کنید.