Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -94,3 +94,27 @@ The contribution documentation is available [here](https://github.com/getlago/la
## License

Lago Python client is distributed under [MIT license](LICENSE).

### Payment list filters

```python
from lago_python_client.models import PaymentFilters

filters: PaymentFilters = {
"payment_status": ["succeeded", "failed"],
"currency": "EUR",
"amount_from": 0,
"amount_to": 9223372036854775807,
"created_at_from": "2026-09-01",
"created_at_to": "2026-09-07",
}
client.payments.find_all(filters)
client.customer_payments.find_all("cust_1", filters)
```

The `PaymentFilters` type documents every accepted option. Enum filters accept a
single string or a list; lists use repeated bracketed query keys. All filters
combine with AND, while values in one list combine with OR. Amount bounds are
inclusive integer cents. Receipt and invoice numbers match exactly, ignoring case.
Date boundaries include the entire day in the organization's timezone. Keep the
same filters when requesting the page number returned in `meta.next_page`.
33 changes: 31 additions & 2 deletions lago_python_client/customers/payments_client.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,13 @@
from typing import ClassVar, Type
from typing import Any, ClassVar, Mapping, Optional, Type

import httpx

from ..base_client import BaseClient
from ..mixins import FindAllChildrenCommandMixin
from ..mixins import DEFAULT_TIMEOUT, FindAllChildrenCommandMixin
from ..models.payment import PaymentResponse
from ..payments.filters import payment_filter_options
from ..services.request import QueryPairs, make_headers, make_url, send_get_request
from ..services.response import get_response_data, prepare_index_response
from .clients import CustomerClient


Expand All @@ -11,3 +16,27 @@ class CustomerPaymentsClient(FindAllChildrenCommandMixin, BaseClient):
API_RESOURCE: ClassVar[str] = "payments"
RESPONSE_MODEL: ClassVar[Type[PaymentResponse]] = PaymentResponse
ROOT_NAME: ClassVar[str] = "payment"

def find_all(
self, resource_id: str, options: QueryPairs = None, timetour: Optional[httpx.Timeout] = None
) -> Mapping[str, Any]:
"""List a customer's payments with PaymentFilters; resource_id supplies external_customer_id.

All other filters and array serialization match PaymentClient.find_all.
Keep the existing positional and keyword spelling of timetour for compatibility.
"""
response = send_get_request(
url=make_url(
origin=self.base_url,
path_parts=(self.PARENT_API_RESOURCE, resource_id, self.API_RESOURCE),
query_pairs=payment_filter_options(options),
),
headers=make_headers(api_key=self.api_key),
timeout=timetour if timetour is not None else DEFAULT_TIMEOUT,
rate_limit_retry_config=self.rate_limit_retry_config,
)
return prepare_index_response(
api_resource=self.API_RESOURCE,
response_model=self.RESPONSE_MODEL,
data=get_response_data(response=response),
)
2 changes: 1 addition & 1 deletion lago_python_client/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,7 @@
from .organization import (
OrganizationBillingConfiguration as OrganizationBillingConfiguration,
)
from .payment import Payment as Payment
from .payment import Payment as Payment, PaymentFilters as PaymentFilters
from .payment_receipt import (
PaymentReceiptResponse as PaymentReceiptResponse,
)
Expand Down
31 changes: 29 additions & 2 deletions lago_python_client/models/payment.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,34 @@
from typing import List, Optional
from typing import List, Literal, Optional, TypedDict, Union

from ..base_model import BaseModel, BaseResponseModel

PaymentStatusFilter = Literal["pending", "processing", "succeeded", "failed"]
PaymentProviderFilter = Literal["stripe", "gocardless", "cashfree", "adyen", "flutterwave", "moneyhash"]
PaymentTypeFilter = Literal["manual", "provider"]
PayableTypeFilter = Literal["Invoice", "PaymentRequest"]


class PaymentFilters(TypedDict, total=False):
"""Optional list filters; amounts are inclusive integer cents and dates are ISO-8601 dates."""

page: int
per_page: int
external_customer_id: str
invoice_id: str
payment_status: Union[PaymentStatusFilter, List[PaymentStatusFilter]]
payment_statuses: Union[PaymentStatusFilter, List[PaymentStatusFilter]]
amount_from: int
amount_to: int
receipt_number: str
created_at_from: str
created_at_to: str
payment_provider_type: Union[PaymentProviderFilter, List[PaymentProviderFilter]]
currency: str
invoice_number: str
payment_type: Union[PaymentTypeFilter, List[PaymentTypeFilter]]
payable_type: Union[PayableTypeFilter, List[PayableTypeFilter]]
search_term: str


class Payment(BaseModel):
invoice_id: str
Expand All @@ -18,7 +45,7 @@ class PaymentResponse(BaseResponseModel):
amount_currency: str
payment_status: str
type: str
reference: str
reference: Optional[str]
external_payment_id: Optional[str]
created_at: str

Expand Down
22 changes: 20 additions & 2 deletions lago_python_client/payments/clients.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,12 @@
from typing import ClassVar, Type
from typing import Any, ClassVar, Mapping, Optional, Type

import httpx

from ..base_client import BaseClient
from ..mixins import CreateCommandMixin, FindAllCommandMixin, FindCommandMixin
from ..mixins import DEFAULT_TIMEOUT, CreateCommandMixin, FindAllCommandMixin, FindCommandMixin
from ..models.payment import PaymentResponse
from ..services.request import QueryPairs
from .filters import payment_filter_options


class PaymentClient(
Expand All @@ -14,3 +18,17 @@ class PaymentClient(
API_RESOURCE: ClassVar[str] = "payments"
RESPONSE_MODEL: ClassVar[Type[PaymentResponse]] = PaymentResponse
ROOT_NAME: ClassVar[str] = "payment"

def find_all(
self, options: QueryPairs = None, timeout: Optional[httpx.Timeout] = DEFAULT_TIMEOUT
) -> Mapping[str, Any]:
"""List payments using PaymentFilters or query pairs.

Accepted keys: page, per_page, external_customer_id, invoice_id, payment_status
(or payment_statuses), amount_from, amount_to, receipt_number, created_at_from,
created_at_to, payment_provider_type, currency, invoice_number, payment_type,
payable_type and search_term.
Enum filters accept a string or list; lists use repeated bracketed query keys.
Amount bounds are inclusive integer cents (0 through 9223372036854775807).
"""
return super().find_all(payment_filter_options(options), timeout)
24 changes: 24 additions & 0 deletions lago_python_client/payments/filters.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
from collections.abc import Mapping

from ..services.request import QueryPairs

PAYMENT_ARRAY_FILTERS = {
"payment_status",
"payment_statuses",
"payment_provider_type",
"payment_type",
"payable_type",
}


def payment_filter_options(options: QueryPairs = None) -> QueryPairs:
"""Encode payment arrays with Rails brackets without mutating the caller's options."""
pairs = options.items() if isinstance(options, Mapping) else options or []
result = []
for key, value in pairs:
if isinstance(value, (list, tuple)):
name = f"{key}[]" if key in PAYMENT_ARRAY_FILTERS else key
result.extend((name, item) for item in value)
else:
result.append((key, value))
return result
100 changes: 100 additions & 0 deletions tests/test_payment_filters.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
from copy import deepcopy
from urllib.parse import parse_qs, urlparse

import httpx
import pytest
from pytest_httpx import HTTPXMock

from lago_python_client.client import Client
from lago_python_client.models import PaymentFilters

from .utils.mixin import mock_response


@pytest.mark.parametrize("customer_scoped", [False, True])
def test_payment_filters_serialize_exactly(httpx_mock: HTTPXMock, customer_scoped):
options: PaymentFilters = {
"page": 2,
"per_page": 5,
"invoice_id": "1a901a90-1a90-1a90-1a90-1a901a901a90",
"payment_status": ["succeeded", "failed"],
"payment_statuses": ["pending"],
"amount_from": 0,
"amount_to": 9223372036854775807,
"receipt_number": "Rcpt & +/#1",
"created_at_from": "2026-09-01",
"created_at_to": "2026-09-07",
"payment_provider_type": ["stripe", "gocardless"],
"currency": "EUR",
"invoice_number": "LAG & +/#2",
"payment_type": ["manual", "provider"],
"payable_type": ["Invoice", "PaymentRequest"],
"search_term": "pi_3 & +/#",
}
if not customer_scoped:
options["external_customer_id"] = "cust_1"
original = deepcopy(options)
client = Client(api_key="test_key")
httpx_mock.add_response(content=mock_response(mock="payment_index"))
timeout = httpx.Timeout(17)
if customer_scoped:
result = client.customer_payments.find_all("cust_1", options, timeout)
else:
result = client.payments.find_all(options, timeout)
request = httpx_mock.get_request()
query = parse_qs(urlparse(str(request.url)).query)
expected = {
"page": ["2"],
"per_page": ["5"],
"invoice_id": ["1a901a90-1a90-1a90-1a90-1a901a901a90"],
"payment_status[]": ["succeeded", "failed"],
"payment_statuses[]": ["pending"],
"amount_from": ["0"],
"amount_to": ["9223372036854775807"],
"receipt_number": ["Rcpt & +/#1"],
"created_at_from": ["2026-09-01"],
"created_at_to": ["2026-09-07"],
"payment_provider_type[]": ["stripe", "gocardless"],
"currency": ["EUR"],
"invoice_number": ["LAG & +/#2"],
"payment_type[]": ["manual", "provider"],
"payable_type[]": ["Invoice", "PaymentRequest"],
"search_term": ["pi_3 & +/#"],
}
if not customer_scoped:
expected["external_customer_id"] = ["cust_1"]
assert query == expected
assert request.url.path == ("/api/v1/customers/cust_1/payments" if customer_scoped else "/api/v1/payments")
assert request.headers["Authorization"] == "Bearer test_key"
assert request.extensions["timeout"]["read"] == 17
assert options == original
assert result["meta"]["current_page"] == 1


@pytest.mark.parametrize(
"options, expected",
[
({"payment_status": "processing"}, {"payment_status": ["processing"]}),
({"payment_status[]": ["succeeded", "failed"]}, {"payment_status[]": ["succeeded", "failed"]}),
(
[("payment_status[]", "succeeded"), ("payment_status[]", "failed")],
{"payment_status[]": ["succeeded", "failed"]},
),
({"payment_type": []}, {}),
],
)
def test_payment_filter_options_remain_compatible(httpx_mock: HTTPXMock, options, expected):
httpx_mock.add_response(content=mock_response(mock="payment_index"))
Client(api_key="test_key").payments.find_all(options)
assert parse_qs(urlparse(str(httpx_mock.get_request().url)).query) == expected


def test_provider_payment_without_reference(httpx_mock: HTTPXMock):
import json

data = json.loads(mock_response(mock="payment_index"))
data["payments"][0]["reference"] = None
data["payments"][0]["type"] = "provider"
httpx_mock.add_response(json=data)
result = Client(api_key="test_key").payments.find_all({"payment_type": ["provider"]})
assert result["payments"][0].reference is None
Loading