epay-sdk

A production-oriented Python SDK for common EPay-style payment gateways.

The project focuses on the full payment lifecycle rather than only assembling order parameters.

Highlights

• MD5 signing and signature verification
• Create-order URL generation and raw gateway submission
• Order query requests and response parsing
• Callback signature verification
• Optional query reconciliation after callbacks
• Atomic, idempotent repository protocol
• Optional SQLAlchemy repository implementation
• Decimal-based amount normalization
• Library-friendly logging defaults
• Django integration template that keeps framework code outside the core package

Installation

uv (recommended)

Core package:

uv add epay-sdk

With the optional SQLAlchemy integration:

uv add 'epay-sdk[sqlalchemy]'

With the Django example dependencies:

uv add 'epay-sdk[django]'

For local development in this repository:

uv sync --extra dev

pip fallback

If you are not using uv, the package is also available on PyPI:

pip install epay-sdk

Quick start

from epay_sdk import EPayClient, EPayConfig, PayType, PaymentService

client = EPayClient(
    EPayConfig(
        pid="1001",
        key="your_secret_key",
        base_url="https://your-epay.com",
        environment="production",
    )
)
service = PaymentService(client)

pay_url = service.create_order(
    pay_type=PayType.ALIPAY,
    order_no="ORDER_10001",
    amount="9.90",
    subject="Membership top-up",
    notify_url="https://api.example.com/pay/notify",
    return_url="https://www.example.com/pay/success",
)
print(pay_url)

PaymentService also keeps convenience helpers:

• create_alipay_order(...)
• create_wxpay_order(...)
• create_qqpay_order(...)

Choosing an order-creation API

• PaymentService.create_order(...) — the recommended business-facing entry point. It normalizes the amount and returns a ready-to-use payment URL.
• EPayClient.create_order_url(CreateOrderRequest) — lower level; use it when you already build CreateOrderRequest yourself.
• EPayClient.create_order(CreateOrderRequest) — sends the request to the gateway and returns the raw upstream response body.

If your use case is simply “generate a payment URL”, prefer PaymentService.create_order(...) or one of its convenience helpers.

Recommended callback flow

A production-friendly callback flow looks like this:

1. Receive the gateway callback form payload.
2. Call client.verify_callback() to verify the signature and validate pid, sign_type, required fields, and amount format.
3. Call CallbackProcessor.process() to load the local order and validate the amount.
4. If verify_with_query=True, the processor performs query_order() + parse_query_result() for upstream reconciliation.
5. Persist the paid transition through mark_paid_if_unpaid().
6. Record callback audit stages through record_callback_attempt(..., stage=...).
7. Let the application decide whether to commit or roll back the surrounding transaction.
8. Return success for accepted callbacks and fail for rejected/retryable callbacks.

Current callback audit stages:

• RECEIVED
• RECONCILED
• APPLIED
• REJECTED

Repository contract

You are expected to implement your own order repository, but it should satisfy this protocol:

class OrderRepository(Protocol):
    def get_order(self, order_no: str) -> Optional[OrderRecord]: ...
    def mark_paid_if_unpaid(self, order_no: str, gateway_trade_no: str, raw_payload: str) -> bool: ...
    def record_callback_attempt(
        self,
        order_no: str,
        accepted: bool,
        reason: Optional[str],
        raw_payload: str,
        *,
        stage: str,
    ) -> None: ...

Important behavior notes:

• mark_paid_if_unpaid() must be atomic, typically via a conditional database update.
• record_callback_attempt() must persist the stage field — it is part of the public contract.
• If your repository supports transaction control, it may additionally expose commit() / rollback().
• CallbackProcessor.process() defaults to not committing or rolling back external transactions.
• If you explicitly want the processor to manage the repository transaction boundary, use CallbackProcessor(..., manage_transaction=True).

Optional SQLAlchemy integration

Install the SQLAlchemy extra with:

uv add 'epay-sdk[sqlalchemy]'

The top-level package lazily exposes these symbols when SQLAlchemy is installed:

• Base
• PaymentOrderModel
• CallbackAuditModel
• SQLAlchemyOrderRepository
• create_sqlite_engine

Example:

from sqlalchemy.orm import Session
from epay_sdk import Base, PaymentOrderModel, SQLAlchemyOrderRepository, create_sqlite_engine

engine = create_sqlite_engine("sqlite:///./epay.db")
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(PaymentOrderModel(order_no="A001", amount="9.90", status="UNPAID"))
    session.commit()

    repo = SQLAlchemyOrderRepository(session)
    if repo.mark_paid_if_unpaid("A001", "G100", '{"trade_no":"G100"}'):
        repo.record_callback_attempt("A001", True, None, '{"trade_no":"G100"}', stage="APPLIED")
        repo.commit()

If SQLAlchemy is not installed, import epay_sdk still works. Only accessing SQLAlchemy-specific exports requires the extra.

Django integration template

The Django strategy is intentionally app-layer oriented:

• src/epay_sdk/ stays framework-agnostic
• Django models, repositories, views, URLs, and transaction boundaries live in the Django app layer
• the repository ships examples/django_app/ as a ready-to-copy integration template

This keeps the SDK generic while still providing a concrete Django reference implementation.

Install the example dependency set with:

uv add 'epay-sdk[django]'

Or, when working from this repository:

uv sync --extra django

Django transaction pattern

For Django, the recommended transaction pattern is:

• keep CallbackProcessor(..., manage_transaction=False) (the default)
• wrap callback processing in transaction.atomic()
• let Django own the commit/rollback behavior

Example:

from django.db import transaction

with transaction.atomic():
    callback = client.verify_callback(request.POST.dict())
    repo = DjangoOrderRepository()
    processor = CallbackProcessor(client, repo, verify_with_query=True)
    result = processor.process(callback)
return HttpResponse(result.response_text, content_type="text/plain")

Mapping the repository contract in Django

In Django terms, the repository usually maps like this:

• get_order(order_no) — load a Django model and map it to OrderRecord
• mark_paid_if_unpaid(order_no, gateway_trade_no, raw_payload) — use a single conditional update
• record_callback_attempt(...) — write a callback audit row

The most important part is keeping mark_paid_if_unpaid() atomic. In Django, that typically means:

updated = PaymentOrder.objects.filter(
    order_no=order_no,
    status="UNPAID",
).update(
    status="PAID",
    gateway_trade_no=gateway_trade_no,
    raw_payload=raw_payload,
    paid_at=timezone.now(),
)
first_success = updated == 1

That mirrors the core SDK requirement that only UNPAID -> PAID is allowed for the idempotent paid transition.

What the Django example includes

examples/django_app/ currently shows:

• Django model definitions
• a DjangoOrderRepository implementation via duck typing
• an order-creation view using PaymentService.create_order(...)
• callback handling through client.verify_callback(...) + CallbackProcessor.process(...)
• transaction.atomic() for transaction control
• app/project URL wiring and Django tests

Security and runtime notes

• The signing protocol follows the common EPay convention: sorted non-empty parameters plus the merchant key, hashed with MD5. This is for gateway compatibility, not a modern standalone transport-security mechanism.
• Use HTTPS in production and keep verify_ssl=True. The SDK rejects verify_ssl=False when environment="production".
• SQLite is fine for local development and examples, but production callback processing is better served by PostgreSQL/MySQL or another database with stronger concurrency semantics.
• Always verify callbacks before processing them.
• The environment field is currently a safety/configuration signal. It influences validation and warnings, but does not automatically switch gateway endpoints.

Examples

• examples/flask_app.py — minimal Flask integration with an in-memory repository
• examples/fastapi_app.py — FastAPI + SQLAlchemy integration with blocking work pushed into the thread pool
• examples/django_app/ — Django integration template that keeps framework-specific code outside the core package

Development

Run the test suite with uv:

uv run pytest tests
uv run python examples/django_app/manage.py test payments

Build release artifacts with:

uv build