Services API

Payment Service

class fastapi_payments.services.payment_service.PaymentService(config, event_publisher, db_session=None)

Bases: object

Service for payment operations.

Methods:

__init__(config, event_publisher[, db_session])	Initialize the payment service.
set_db_session(session)	Set the database session.
get_provider([provider_name])	Get a payment provider instance.
create_customer(email[, name, meta_info, ...])	Create a customer.
ensure_provider_customer(customer_id, provider)	Ensure the customer is registered with the requested provider.
get_customer(customer_id)	Get customer details.
list_customers(*[, limit, offset, search])	List customers stored in the service database.
update_customer(customer_id, **kwargs)	Update customer details.
create_payment_method(customer_id, ...[, ...])	Create a payment method for a customer.
create_setup_intent(customer_id[, provider, ...])	Create a SetupIntent (or provider equivalent) for the given customer so a client-side flow (3DS) can be completed before saving a payment method.
list_payment_methods(customer_id[, provider])	List payment methods for a customer.
update_payment_method(customer_id, method_id)	Update a stored payment method's attributes.
delete_payment_method(customer_id, method_id)	Delete (detach) a stored payment method both in provider (if supported) and DB.
set_default_payment_method(customer_id, ...)	Mark a stored payment method as the default for the customer.
create_product(name[, description, ...])	Create a new product.
list_products(*[, limit, offset])	Return products stored in the local database.
create_plan(product_id, name, pricing_model, ...)	Create a plan for a product.
list_plans(*[, product_id, limit, offset])	Return plans, optionally filtered by product.
create_subscription(customer_id, plan_id[, ...])	Create a subscription for a customer.
get_subscription(subscription_id)	Get subscription details.
list_subscriptions(*[, customer_id, status, ...])	Return subscriptions filtered by optional criteria.
cancel_subscription(subscription_id[, ...])	Cancel a subscription.
process_payment(customer_id, amount, currency)	Process a one-time payment.
sync_resources([resources, provider, ...])	Synchronize local DB state with provider(s).
create_sync_job([resources, provider, filters])	Create a SyncJob record and return its basic details.
execute_sync_job(job_id)	Run sync_resources for a previously created job and persist results.
refund_payment(payment_id[, amount])	Refund a payment, partially or fully.
list_payments(*[, customer_id, status, ...])	Return payments filtered by optional criteria.
record_usage(subscription_id, quantity[, ...])	Record usage for a subscription.
handle_webhook(provider, payload[, signature])	Handle webhooks from payment providers.

Parameters:

• config (PaymentConfig)

• event_publisher (PaymentEventPublisher)

__init__(config, event_publisher, db_session=None)

Initialize the payment service.

Parameters:

• config (PaymentConfig) – Payment configuration

• event_publisher (PaymentEventPublisher) – Event publisher for notifications

• db_session – Database session

set_db_session(session)

Set the database session.

Parameters:

session (AsyncSession) – SQLAlchemy AsyncSession

get_provider(provider_name=None)

Get a payment provider instance.

Parameters:

provider_name (Optional[str]) – Name of the provider to get, or None for default

Return type:

Any

Returns:

Provider instance

async create_customer(email, name=None, meta_info=None, provider=None, address=None)

Create a customer.

Parameters:

• email (str) – Customer email

• name (Optional[str]) – Customer name

• meta_info (Optional[Dict[str, Any]]) – Additional customer meta_info

• provider (Optional[str]) – Name of the provider to use (default is default_provider)

• address (Dict[str, Any] | None)

Return type:

Dict[str, Any]

Returns:

Customer data dictionary

async ensure_provider_customer(customer_id, provider)

Ensure the customer is registered with the requested provider.

Return type:

Dict[str, Any]

Parameters:

• customer_id (str)

• provider (str)

async get_customer(customer_id)

Get customer details.

Parameters:

customer_id (str) – Customer ID

Return type:

Optional[Dict[str, Any]]

Returns:

Customer data if found, None otherwise

async list_customers(*, limit=50, offset=0, search=None)

List customers stored in the service database.

Return type:

List[Dict[str, Any]]

Parameters:

• limit (int)

• offset (int)

• search (str | None)

async update_customer(customer_id, **kwargs)

Update customer details.

Parameters:

• customer_id (str) – Customer ID

• **kwargs – Fields to update

Return type:

Dict[str, Any]

Returns:

Updated customer data

async create_payment_method(customer_id, payment_details, provider=None)

Create a payment method for a customer.

Parameters:

• customer_id (str) – Customer ID

• payment_details (Dict[str, Any]) – Payment method details

• provider (Optional[str]) – Optional provider name

Return type:

Dict[str, Any]

Returns:

Created payment method data

async create_setup_intent(customer_id, provider=None, usage=None)

Create a SetupIntent (or provider equivalent) for the given customer so a client-side flow (3DS) can be completed before saving a payment method.

Returns a dict that includes id and client_secret keys when supported by the provider.

Return type:

Dict[str, Any]

Parameters:

• customer_id (str)

• provider (str | None)

• usage (str | None)

async list_payment_methods(customer_id, provider=None)

List payment methods for a customer.

Parameters:

• customer_id (str) – Customer ID

• provider (Optional[str]) – Optional provider name

Return type:

List[Dict[str, Any]]

Returns:

List of payment methods

async update_payment_method(customer_id, method_id, provider=None, **fields)

Update a stored payment method’s attributes.

Supported updates include setting is_default, or meta_info.

Return type:

Dict[str, Any]

Parameters:

• customer_id (str)

• method_id (str)

• provider (str | None)

• fields (Any)

async delete_payment_method(customer_id, method_id, provider=None)

Delete (detach) a stored payment method both in provider (if supported) and DB.

Return type:

Dict[str, Any]

Parameters:

• customer_id (str)

• method_id (str)

• provider (str | None)

async set_default_payment_method(customer_id, method_id, provider=None)

Mark a stored payment method as the default for the customer.

Updates both DB (via repository.set_default) and attempts to update the provider’s customer default payment settings where supported.

Return type:

Dict[str, Any]

Parameters:

• customer_id (str)

• method_id (str)

• provider (str | None)

async create_product(name, description=None, provider=None, meta_info=None)

Create a new product.

Parameters:

• name (str) – Product name

• description (Optional[str]) – Optional product description

• provider (Optional[str]) – Optional provider name

• meta_info (Optional[Dict[str, Any]]) – Optional product meta_info

Return type:

Dict[str, Any]

Returns:

Created product data

async list_products(*, limit=50, offset=0)

Return products stored in the local database.

Return type:

List[Dict[str, Any]]

Parameters:

• limit (int)

• offset (int)

async create_plan(product_id, name, pricing_model, amount, description=None, currency='USD', billing_interval=None, billing_interval_count=None, trial_period_days=None, provider=None, meta_info=None)

Create a plan for a product.

Parameters:

• product_id (str) – Product ID

• name (str) – Plan name

• pricing_model (str) – Pricing model (subscription, usage_based, etc.)

• amount (float) – Base amount

• description (Optional[str]) – Optional plan description

• currency (str) – Currency code

• billing_interval (Optional[str]) – Billing interval (day, week, month, year)

• billing_interval_count (Optional[int]) – Number of intervals between billings

• trial_period_days (Optional[int]) – Optional trial period in days

• provider (Optional[str]) – Optional provider name

• meta_info (Optional[Dict[str, Any]]) – Optional plan meta_info

Return type:

Dict[str, Any]

Returns:

Created plan data

async list_plans(*, product_id=None, limit=50, offset=0)

Return plans, optionally filtered by product.

Return type:

List[Dict[str, Any]]

Parameters:

• product_id (str | None)

• limit (int)

• offset (int)

async create_subscription(customer_id, plan_id, quantity=1, trial_period_days=None, meta_info=None)

Create a subscription for a customer.

Parameters:

• customer_id (str) – Customer ID

• plan_id (str) – Plan ID

• quantity (int) – Number of units/seats

• trial_period_days (Optional[int]) – Optional trial period in days

• meta_info (Optional[Dict[str, Any]]) – Optional subscription meta_info

Return type:

Dict[str, Any]

Returns:

Created subscription data

async get_subscription(subscription_id)

Get subscription details.

Parameters:

subscription_id (str) – Subscription ID

Return type:

Optional[Dict[str, Any]]

Returns:

Subscription data if found, None otherwise

async list_subscriptions(*, customer_id=None, status=None, limit=50, offset=0)

Return subscriptions filtered by optional criteria.

Return type:

List[Dict[str, Any]]

Parameters:

• customer_id (str | None)

• status (str | None)

• limit (int)

• offset (int)

async cancel_subscription(subscription_id, cancel_at_period_end=True)

Cancel a subscription.

Parameters:

• subscription_id (str) – Subscription ID

• cancel_at_period_end (bool) – Whether to cancel at the end of the current period

Return type:

Dict[str, Any]

Returns:

Updated subscription data

async process_payment(customer_id, amount, currency, payment_method_id=None, description=None, meta_info=None, mandate_id=None, provider=None)

Process a one-time payment.

Parameters:

• customer_id (str) – Customer ID

• amount (float) – Payment amount

• currency (str) – Currency code

• payment_method_id (Optional[str]) – Optional payment method ID

• description (Optional[str]) – Optional payment description

• meta_info (Optional[Dict[str, Any]]) – Optional payment meta_info

• mandate_id (str | None)

• provider (str | None)

Return type:

Dict[str, Any]

Returns:

Payment data

async sync_resources(resources=None, provider=None, filters=None, limit=100, offset=0)

Synchronize local DB state with provider(s).

This is best-effort: for each requested resource we attempt to fetch authoritative data from the configured provider(s) and update local rows where possible. If a provider does not implement the expected retrieval method for a resource, the resource is skipped for that provider.

Parameters:

• resources (Optional[List[str]]) – list of resource names to sync (customers, products, plans, payments, subscriptions, payment_methods). If None, all supported resources are synced.

• provider (Optional[str]) – optional provider name to limit sync to one provider.

• filters (Optional[Dict[str, Any]]) – resource-specific filters e.g. {‘customer_id’: ‘…’}

• limit/offset – pagination for resource listing.

• limit (int)

• offset (int)

Return type:

Dict[str, Any]

Returns:

A dict summarizing how many items were examined/updated and any errors encountered for each resource.

async create_sync_job(resources=None, provider=None, filters=None)

Create a SyncJob record and return its basic details.

Actual work is performed by execute_sync_job which is expected to be scheduled as a background task by the API layer.

Return type:

Dict[str, Any]

Parameters:

• resources (List[str] | None)

• provider (str | None)

• filters (Dict[str, Any] | None)

async execute_sync_job(job_id)

Run sync_resources for a previously created job and persist results.

This method will set job status to ‘running’, invoke sync_resources, and finally store the result and mark job ‘completed’ or ‘failed’.

Parameters:

job_id (str)

async refund_payment(payment_id, amount=None)

Refund a payment, partially or fully.

Parameters:

• payment_id (str) – Payment ID

• amount (Optional[float]) – Optional refund amount (full refund if not specified)

Return type:

Dict[str, Any]

Returns:

Updated payment data

async list_payments(*, customer_id=None, status=None, limit=50, offset=0)

Return payments filtered by optional criteria.

Return type:

List[Dict[str, Any]]

Parameters:

• customer_id (str | None)

• status (str | None)

• limit (int)

• offset (int)

async record_usage(subscription_id, quantity, timestamp=None, description=None)

Record usage for a subscription.

Parameters:

• subscription_id (str) – Subscription ID

• quantity (float) – Usage quantity

• timestamp (Optional[str]) – Optional usage timestamp

• description (Optional[str]) – Optional usage description

Return type:

Dict[str, Any]

Returns:

Usage record data

async handle_webhook(provider, payload, signature=None)

Handle webhooks from payment providers.

Parameters:

• provider (str) – Provider name

• payload (Any) – Webhook payload

• signature (Optional[str]) – Optional webhook signature

Return type:

Dict[str, Any]

Returns:

Processed webhook data

Pricing Service

Dependency Injection

fastapi_payments.services.payment_service.get_payment_service()

Dependency to get payment service instance.