Skip to main content

FastAPI Integration

Complete guide for integrating PayTechUZ with FastAPI applications.

Complete Example Project

Looking for a ready-to-use FastAPI project? Check out our FastAPI Example Repository with full payment integration!

Installation​

Install PayTechUZ with FastAPI support:

pip install paytechuz[fastapi]

Configuration​

Set up database models:

from datetime import datetime, timezone

from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey

from paytechuz.integrations.fastapi import Base as PaymentsBase
from paytechuz.integrations.fastapi.models import run_migrations

# Create database engine
SQLALCHEMY_DATABASE_URL = "sqlite:///./payments.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)

# Create base declarative class
Base = declarative_base()

# Create Order model
class Order(Base):
    __tablename__ = "orders"

    id = Column(Integer, primary_key=True, index=True)
    product_name = Column(String, index=True)
    total_amount = Column(Float)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))

# Create Invoice model
class Invoice(Base):
    __tablename__ = "invoices"

    id = Column(Integer, primary_key=True, index=True)
    order_id = Column(Integer, ForeignKey("orders.id"))
    amount = Column(Float)
    status = Column(String, default="pending")
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))

    order = relationship("Order")

# Create payment tables using run_migrations
run_migrations(engine)

# Create Order and Invoice tables
Base.metadata.create_all(bind=engine)

# Create session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Create webhook handlers:​

from fastapi import FastAPI, Request, Depends
from sqlalchemy.orm import Session
from paytechuz.integrations.fastapi import (
    PaymeWebhookHandler, 
    ClickWebhookHandler, 
    AtmosWebhookHandler
)

app = FastAPI()

# Dependency to get the database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

class CustomPaymeWebhookHandler(PaymeWebhookHandler):
    def successfully_payment(self, params, transaction):
        invoice = self.db.query(Invoice).filter(Invoice.id == transaction.account_id).first()
        invoice.status = "paid"
        self.db.commit()

    def cancelled_payment(self, params, transaction):
        invoice = self.db.query(Invoice).filter(Invoice.id == transaction.account_id).first()
        invoice.status = "cancelled"
        self.db.commit()

class CustomClickWebhookHandler(ClickWebhookHandler):
    def successfully_payment(self, params, transaction):
        invoice = self.db.query(Invoice).filter(Invoice.id == transaction.account_id).first()
        invoice.status = "paid"
        self.db.commit()

    def cancelled_payment(self, params, transaction):
        invoice = self.db.query(Invoice).filter(Invoice.id == transaction.account_id).first()
        invoice.status = "cancelled"
        self.db.commit()

class CustomAtmosWebhookHandler(AtmosWebhookHandler):
    def successfully_payment(self, params, transaction):
        invoice = self.db.query(Invoice).filter(Invoice.id == transaction.account_id).first()
        invoice.status = "paid"
        self.db.commit()

    def cancelled_payment(self, params, transaction):
        invoice = self.db.query(Invoice).filter(Invoice.id == transaction.account_id).first()
        invoice.status = "cancelled"
        self.db.commit()

@app.post("/payments/payme/webhook")
async def payme_webhook(request: Request, db: Session = Depends(get_db)):
    handler = CustomPaymeWebhookHandler(
        db=db,
        payme_id="your_merchant_id",
        payme_key="your_merchant_key",
        account_model=Invoice,
        account_field='id',
        amount_field='amount'
    )
    return await handler.handle_webhook(request)

@app.post("/payments/click/webhook")
async def click_webhook(request: Request, db: Session = Depends(get_db)):
    handler = CustomClickWebhookHandler(
        db=db,
        service_id="your_service_id",
        secret_key="your_secret_key",
        account_model=Invoice
    )
    return await handler.handle_webhook(request)

@app.post("/payments/atmos/webhook")
async def atmos_webhook(request: Request, db: Session = Depends(get_db)):
    handler = CustomAtmosWebhookHandler(
        db=db,
        consumer_key="your_consumer_key",
        consumer_secret="your_consumer_secret",
        store_id="your_store_id",
        api_key="your_api_key",
        account_model=Invoice,
        account_field='id'
    )
    return await handler.handle_webhook(request)
import os
from pydantic import BaseModel
from paytechuz.gateways.payme import PaymeGateway
from paytechuz.gateways.click import ClickGateway
from paytechuz.gateways.atmos import AtmosGateway

class OrderCreate(BaseModel):
    product_name: str
    amount: float
    payment_type: str  # 'payme', 'click', or 'atmos'

@app.post("/order/create")
def create_order(order_data: OrderCreate, db: Session = Depends(get_db)):
    # 1. Create Order
    order = Order(
        product_name=order_data.product_name, 
        total_amount=order_data.amount
    )
    db.add(order)
    db.commit()
    db.refresh(order)

    # 2. Create Invoice linked to the Order
    invoice = Invoice(
        order_id=order.id, 
        amount=order.total_amount,
        status="pending"
    )
    db.add(invoice)
    db.commit()
    db.refresh(invoice)

    payment_url = None

    # 3. Generate Payment Link based on payment_type
    if order_data.payment_type == 'payme':
        gateway = PaymeGateway(
            payme_id='your_payme_id',
            payme_key='your_payme_key',
            is_test_mode=True
        )
        payment_url = gateway.create_payment(
            id=invoice.id,
            amount=int(invoice.amount * 100),  # Payme works with tiyin
            return_url="https://example.com/success"
        )

    elif order_data.payment_type == 'click':
        gateway = ClickGateway(
            service_id='your_click_service_id',
            merchant_id='your_click_merchant_id',
            merchant_user_id='your_click_merchant_user_id',
            secret_key='your_click_secret_key',
            is_test_mode=True
        )
        payment_url = gateway.create_payment(
            id=invoice.id,
            amount=invoice.amount,
            return_url="https://example.com/success"
        )

    elif order_data.payment_type == 'atmos':
        gateway = AtmosGateway(
            consumer_key='your_atmos_consumer_key',
            consumer_secret='your_atmos_consumer_secret',
            store_id='your_atmos_store_id',
            is_test_mode=True
        )
        payment_data = gateway.create_payment(
            account_id=invoice.id,
            amount=int(invoice.amount * 100)  # Atmos works with tiyin
        )
        payment_url = payment_data['payment_url']

    return {
        "order_id": order.id, 
        "invoice_id": invoice.id, 
        "payment_url": payment_url
    }

Next Steps​