Modules

Top-level package for Setu UPI DeepLinks SDK.

`auth` ¶

Auth module.

`generate_jwt_token(scheme_id, secret)` ¶

Generate a JWT token.

Parameters:

Name	Type	Description	Default
`scheme_id`	`str`	description	required
`secret`	`str`	description	required

Returns:

Type	Description
`str`	description

Source code in setu/auth.py

def generate_jwt_token(scheme_id: str, secret: str) -> str:
    """Generate a JWT token.

    Args:
        scheme_id (str): _description_
        secret (str): _description_

    Returns:
        str: _description_
    """
    payload = {
        "aud": scheme_id,
        "iat": datetime.datetime.utcnow(),
        "jti": str(uuid.uuid1()),
    }

    return "Bearer {}".format(jwt.encode(payload, secret, algorithm="HS256"))

`generate_oauth_token(client_id, secret, mode='SANDBOX')` ¶

Generate an OAuth token.

Parameters:

Name	Type	Description	Default
`client_id`	`str`	description	required
`secret`	`str`	description	required
`mode`	`Mode`	description. Defaults to "SANDBOX".	`'SANDBOX'`

Returns:

Type	Description
`str`	description

Source code in setu/auth.py

def generate_oauth_token(
    client_id: str,
    secret: str,
    mode: Mode = "SANDBOX",
) -> str:
    """Generate an OAuth token.

    Args:
        client_id (str): _description_
        secret (str): _description_
        mode (Mode, optional): _description_. Defaults to "SANDBOX".

    Returns:
        str: _description_
    """
    payload = {
        "clientID": client_id,
        "secret": secret,
    }
    headers = {"Content-Type": "application/json"}
    response = requests.post(
        get_url_path(API.FETCH_TOKEN, AUTH_TYPE_OAUTH, mode),
        json=payload,
        headers=headers,
    )

    if response.status_code == 200:
        data = response.json()
        return "Bearer {}".format(data["data"]["token"])
    raise

`body` ¶

Body builder helper module.

`get_batch_refund_body(**kwargs)` ¶

Get the payload for Initiate Batch Refund API.

Source code in setu/body.py

def get_batch_refund_body(**kwargs) -> Dict[str, Any]:
    """Get the payload for Initiate Batch Refund API."""
    payload = {
        "refunds": [
            {
                "seqNo": i,
                "identifier": refund.identifier,
                "identifierType": refund.identifierType,
                "refundType": refund.refundType,
                "refundAmount": refund.refundAmount,
                "deductions": [
                    {
                        "account": {"id": deduction.account_number, "ifsc": deduction.account_ifsc},
                        "split": {"value": deduction.amount_value, "unit": "INR"},
                    }
                    for deduction in (refund.deductions if refund.deductions is not None else [])
                ],
            }
            for i, refund in enumerate(cast(List[RefundRequestItem], kwargs['refunds']))
        ],
    }

    return payload

`get_create_payment_link_body(**kwargs)` ¶

Get the payload for Create Payment Link API.

Source code in setu/body.py

def get_create_payment_link_body(**kwargs) -> Dict[str, Any]:
    """Get the payload for Create Payment Link API."""
    payload: Dict[str, Any] = {
        "amount": {"currencyCode": "INR", "value": kwargs['amount_value']},
        "amountExactness": kwargs['amount_exactness'],
        "billerBillID": kwargs['biller_bill_id'],
    }

    if kwargs['payee_name'] is not None:
        payload.update({"name": kwargs['payee_name']})

    if kwargs['transaction_note'] is not None:
        payload.update({"transactionNote": kwargs['transaction_note']})

    if kwargs['expiry_date'] is not None:
        expiry_date_str = kwargs['expiry_date'].replace(microsecond=0).isoformat() + "Z"
        payload.update({"expiryDate": expiry_date_str})

    if kwargs['settlement'] is not None:
        payload.update(
            {
                "settlement": {
                    "parts": [
                        {
                            "account": {
                                "id": part.account_number,
                                "ifsc": part.account_ifsc,
                            },
                            "split": {
                                "unit": "INR",
                                "value": part.amount_value,
                            },
                        }
                        for part in kwargs['settlement'].parts
                    ],
                    "primaryAccount": {
                        "id": kwargs['settlement'].primary_account.account_number,
                        "ifsc": kwargs['settlement'].primary_account.account_ifsc,
                    },
                }
            }
        )

    if kwargs['validation_rules'] is not None:
        vr: Dict[str, Any] = {}
        if kwargs['validation_rules'].amount_validation is not None:
            vr.update(
                {
                    "amount": {
                        "maximum": kwargs['validation_rules'].amount_validation.maximum,
                        "minimum": kwargs['validation_rules'].amount_validation.minimum,
                    }
                }
            )
        if kwargs['validation_rules'].source_accounts is not None:
            vr.update(
                {
                    "sourceAccounts": [
                        {"number": source_account.account_number, "ifsc": source_account.account_ifsc}
                        for source_account in kwargs['validation_rules'].source_accounts
                    ]
                }
            )
        payload.update({"validationRules": vr})

    if kwargs['additional_info'] is not None:
        payload.update({"additionalInfo": kwargs['additional_info']})

    if kwargs['campaign_id'] is not None:
        payload.update({"campaignID": kwargs['campaign_id']})

    return payload

`get_mock_credit_body(**kwargs)` ¶

Get the payload for Mock Credit API.

Source code in setu/body.py

def get_mock_credit_body(**kwargs) -> Dict[str, Any]:
    """Get the payload for Mock Credit API."""
    payload = {
        "amount": kwargs['amount_value'],
        "type": "UPI",
        "sourceAccount": {"accountID": "customer@vpa"},
        "destinationAccount": {"accountID": kwargs['upi_id']},
        "transactionReference": kwargs["platform_bill_id"],
    }

    return payload

`get_mock_settlement_body(**kwargs)` ¶

Get the payload for Mock Settlement API.

Source code in setu/body.py

def get_mock_settlement_body(**kwargs) -> Dict[str, Any]:
    """Get the payload for Mock Settlement API."""
    transactions = [{"utr": utr} for utr in kwargs['utrs']]
    payload = {"transactions": transactions}

    return payload

`contract` ¶

Contract module.

`API (Enum)` ¶

Enum containing all the API paths supported by SDK.

Source code in setu/contract.py

class API(Enum):
    """Enum containing all the API paths supported by SDK."""

    FETCH_TOKEN = "/auth/token"
    PAYMENT_LINK_BASE = "/payment-links"
    TRIGGER_MOCK_PAYMENT = "/triggers/funds/addCredit"
    TRIGGER_MOCK_SETTLEMENT = "/triggers/funds/mockSettlement"
    EXPIRE_BILL = "/utilities/bills/{}/expire"
    REPORTS_BASE = "/reports"
    REFUNDS_BASE = "/refund"

`APICredentials (tuple)` ¶

API Credential information.

Source code in setu/contract.py

class APICredentials(NamedTuple):
    """API Credential information."""

    scheme_id: str
    secret: str

`getnewargs(self)` `special` ¶

Return self as a plain tuple. Used by copy and pickle.

Source code in setu/contract.py

def __getnewargs__(self):
    'Return self as a plain tuple.  Used by copy and pickle.'
    return _tuple(self)

`new(_cls, scheme_id, secret)` `special` `staticmethod` ¶

Create new instance of APICredentials(scheme_id, secret)

`repr(self)` `special` ¶

Return a nicely formatted representation string

Source code in setu/contract.py

def __repr__(self):
    'Return a nicely formatted representation string'
    return self.__class__.__name__ + repr_fmt % self

`Account` `dataclass` ¶

Bank account information.

Source code in setu/contract.py

class Account:
    """Bank account information."""

    account_number: str
    account_ifsc: str

`Amount` `dataclass` ¶

Amount information.

Source code in setu/contract.py

class Amount:
    """Amount information."""

    currency_code: str
    value: int

`AmountValidationRule` `dataclass` ¶

Amount validation rules, needed when amount_exactness is RANGE.

Source code in setu/contract.py

class AmountValidationRule:
    """Amount validation rules, needed when amount_exactness is `RANGE`."""

    minimum: int
    maximum: int

`CreatePaymentLinkResponseData` `dataclass` ¶

Response from Create Payment Link API.

Source code in setu/contract.py

class CreatePaymentLinkResponseData:
    """Response from Create Payment Link API."""

    name: str
    payment_link: PaymentLink
    platform_bill_id: str
    campaign_id: Optional[str] = None

`Deduction` `dataclass` ¶

Deduction detail.

Source code in setu/contract.py

class Deduction:
    """Deduction detail."""

    account: Account
    split: Amount

`InitiateBatchRefundResponse` `dataclass` ¶

Initiate Batch Refund Response.

Source code in setu/contract.py

class InitiateBatchRefundResponse:
    """Initiate Batch Refund Response."""

    batch_id: str
    refunds: List[Union[RefundResponseItem, SetuErrorResponseData]]

`MockCreditResponseData` `dataclass` ¶

Mock Credit response.

Source code in setu/contract.py

class MockCreditResponseData:
    """Mock Credit response."""

    utr: str

`PaymentLink` `dataclass` ¶

Payment Link information.

Source code in setu/contract.py

class PaymentLink:
    """Payment Link information."""

    upi_id: str
    upi_link: str
    short_url: Optional[str] = None

`PaymentLinkStatusResponseData` `dataclass` ¶

Response from Payment Link Status API.

Source code in setu/contract.py

class PaymentLinkStatusResponseData:
    """Response from Payment Link Status API."""

    created_at: datetime
    expires_at: datetime
    name: str
    payment_link: PaymentLink
    platform_bill_id: str
    biller_bill_id: str
    status: str
    transaction_note: str
    campaign_id: Optional[str] = None
    payer_vpa: Optional[str] = None
    receipt: Optional[Receipt] = None
    amount_paid: Optional[int] = None
    additional_info: Optional[Dict[str, str]] = None

`Receipt` `dataclass` ¶

Payment receipt information.

Source code in setu/contract.py

class Receipt:
    """Payment receipt information."""

    date: datetime
    id: str

`RefundRequestItem` `dataclass` ¶

Refund request item.

Source code in setu/contract.py

class RefundRequestItem:
    """Refund request item."""

    identifier: str
    identifierType: str
    refundType: str
    refundAmount: Optional[int] = None
    deductions: Optional[List[SplitAccount]] = None

`RefundResponseItem` `dataclass` ¶

Refund response item.

Source code in setu/contract.py

class RefundResponseItem:
    """Refund response item."""

    id: str
    bill_id: str
    transaction_ref_id: str
    amount: Amount
    type: str
    status: str
    created_at: datetime
    deductions: Optional[List[Deduction]] = None
    initiated_at: Optional[datetime] = None

`RefundStatusByIdentifierResponse` `dataclass` ¶

Refund Status By Identifier Response.

Source code in setu/contract.py

class RefundStatusByIdentifierResponse:
    """Refund Status By Identifier Response."""

    refunds: List[RefundResponseItem]

`SettlementSplits` `dataclass` ¶

Settlement split instruction for bill.

Source code in setu/contract.py

class SettlementSplits:
    """Settlement split instruction for bill."""

    parts: List[SplitAccount]
    primary_account: Account

`SetuAPIException (Exception)` `dataclass` ¶

Setu exception class.

Source code in setu/contract.py

class SetuAPIException(Exception):
    """Setu exception class."""

    error: SetuErrorResponseData

`SetuErrorResponseData` `dataclass` ¶

Setu error response data.

Source code in setu/contract.py

class SetuErrorResponseData:
    """Setu error response data."""

    code: str
    detail: str
    title: str
    doc_url: Optional[str] = None
    errors: Optional[List[Any]] = None
    trace_id: Optional[str] = None

`SetuResponseBase` `dataclass` ¶

Setu base response.

Source code in setu/contract.py

class SetuResponseBase:
    """Setu base response."""

    success: bool
    status: int

`SplitAccount (Account)` `dataclass` ¶

Individual settlement split instruction.

Source code in setu/contract.py

class SplitAccount(Account):
    """Individual settlement split instruction."""

    amount_value: int

`ValidationRules` `dataclass` ¶

Validation rules for bill.

Source code in setu/contract.py

class ValidationRules:
    """Validation rules for bill."""

    amount_validation: Optional[AmountValidationRule] = None
    source_accounts: Optional[List[Account]] = None

`deeplink` ¶

Main module.

`Deeplink` ¶

The Deeplink class.

Source code in setu/deeplink.py

class Deeplink:
    """The Deeplink class."""

    def __init__(
        self,
        scheme_id: str,
        secret: str,
        product_instance_id: str,
        auth_type: AuthType = "JWT",
        mode: Mode = "SANDBOX",
    ):
        """Constructor for the Deeplink class.

        Args:
            scheme_id (str): _description_
            secret (str): _description_
            product_instance_id (str): _description_
            auth_type (Literal["JWT", "OAUTH"], optional): _description_. Defaults to "JWT".
            mode (Literal["SANDBOX", "PRODUCTION"], optional): _description_. Defaults to "SANDBOX".
        """
        self.scheme_id = scheme_id
        self.secret = secret
        self.mode = mode
        self.auth_type = auth_type

        self.headers = {
            "X-Setu-Product-Instance-ID": product_instance_id,
            "Content-Type": "application/json",
        }

        self.session = requests.Session()
        self.session.hooks = {"response": lambda r, *args, **kwargs: self.exception_handler(r)}

    def regenerate_token(self):
        """Re-generate token."""
        if self.auth_type == AUTH_TYPE_JWT:
            authorization = generate_jwt_token(self.scheme_id, self.secret)
        else:
            authorization = generate_oauth_token(self.scheme_id, self.secret, self.mode)
        self.headers["Authorization"] = authorization

    @staticmethod
    def exception_handler(r: Response):
        """Exception handler."""
        try:
            r.raise_for_status()
        except requests.exceptions.RequestException as e:
            if isinstance(e, requests.exceptions.HTTPError) and r.reason == "Forbidden":
                raise e
            else:
                error_response_schema = SetuErrorResponseSchema()
                raise SetuAPIException(error_response_schema.load(r.json()))

    class Decorators:
        """Decorators."""

        @staticmethod
        def auth_handler(decorated: Callable):
            """Refresh token."""

            def wrapper(deeplink, *args, **kwargs):
                try:
                    if deeplink.auth_type == AUTH_TYPE_JWT:
                        deeplink.regenerate_token()
                    return decorated(deeplink, *args, **kwargs)
                except requests.exceptions.HTTPError:
                    deeplink.regenerate_token()
                    return decorated(deeplink, *args, **kwargs)

            return wrapper

    @Decorators.auth_handler
    def create_payment_link(
        self,
        amount_value: int,
        biller_bill_id: str,
        amount_exactness: str,
        payee_name: str = None,
        transaction_note: str = None,
        expiry_date: datetime = None,
        settlement: SettlementSplits = None,
        validation_rules: ValidationRules = None,
        campaign_id: str = None,
        additional_info: Dict[str, str] = None,
    ) -> CreatePaymentLinkResponseData:
        """Generate UPI payment link."""
        payload: Dict[str, Any] = get_create_payment_link_body(
            amount_value=amount_value,
            biller_bill_id=biller_bill_id,
            amount_exactness=amount_exactness,
            payee_name=payee_name,
            transaction_note=transaction_note,
            expiry_date=expiry_date,
            settlement=settlement,
            validation_rules=validation_rules,
            campaign_id=campaign_id,
            additional_info=additional_info,
        )

        api_response = self.session.post(
            get_url_path(API.PAYMENT_LINK_BASE, self.auth_type, self.mode),
            json=payload,
            headers=self.headers,
        )
        create_payment_link_response_data_schema = CreatePaymentLinkResponseDataSchema()
        return create_payment_link_response_data_schema.load(api_response.json()['data'])

    @Decorators.auth_handler
    def check_payment_status(self, platform_bill_id: str) -> PaymentLinkStatusResponseData:
        """Check status of UPI payment link."""
        api_response = self.session.get(
            "{}/{}".format(
                get_url_path(API.PAYMENT_LINK_BASE, self.auth_type, self.mode),
                platform_bill_id,
            ),
            headers=self.headers,
        )
        payment_link_status_response_data_schema = PaymentLinkStatusResponseDataSchema()
        return payment_link_status_response_data_schema.load(api_response.json()['data'])

    @Decorators.auth_handler
    def trigger_mock_payment(self, amount_value: float, upi_id: str, platform_bill_id: str) -> MockCreditResponseData:
        """Trigger mock payment for UPI payment link.

        This API is available only on SANDBOX mode.
        """
        if self.mode == MODE_PRODUCTION:
            raise Exception("trigger_mock_payment METHOD IS IS NOT AVAILABLE IN PRODUCTION")

        payload: Dict[str, Any] = get_mock_credit_body(
            amount_value=amount_value, upi_id=upi_id, platform_bill_id=platform_bill_id
        )

        api_response = self.session.post(
            get_url_path(API.TRIGGER_MOCK_PAYMENT, self.auth_type, self.mode),
            json=payload,
            headers=self.headers,
        )
        mock_credit_response_data_schema = MockCreditResponseDataSchema()
        return mock_credit_response_data_schema.load(api_response.json()['data'])

    @Decorators.auth_handler
    def expire_payment_link(self, platform_bill_id: str):
        """Expire payment link."""
        self.session.post(
            get_url_path(API.EXPIRE_BILL, self.auth_type, self.mode).format(platform_bill_id),
            headers=self.headers,
        )
        return

    @Decorators.auth_handler
    def trigger_mock_settlement(self, utrs: List[str]):
        """Trigger mock settlement."""
        payload: Dict[str, Any] = get_mock_settlement_body(utrs=utrs)

        self.session.post(
            get_url_path(API.TRIGGER_MOCK_SETTLEMENT, self.auth_type, self.mode),
            json=payload,
            headers=self.headers,
        )
        return

    @Decorators.auth_handler
    def initiate_batch_refund(
        self,
        refunds: List[RefundRequestItem],
    ) -> InitiateBatchRefundResponse:
        """Initiate batch refund."""
        payload: Dict[str, Any] = get_batch_refund_body(refunds=refunds)

        api_response = self.session.post(
            "{}/batch".format(get_url_path(API.REFUNDS_BASE, self.auth_type, self.mode)),
            json=payload,
            headers=self.headers,
        )
        initiate_batch_refund_response_data_schema = InitiateBatchRefundResponseDataSchema()
        return initiate_batch_refund_response_data_schema.load(api_response.json()['data'])

    @deprecated(version="1.2.0", reason="Use the more generic get_refund_status_by_identifier method instead")
    @Decorators.auth_handler
    def get_batch_refund_status(self, batch_refund_id: str) -> RefundStatusByIdentifierResponse:
        """Get batch refund status."""
        api_response = self.session.get(
            "{}/batch/{}".format(get_url_path(API.REFUNDS_BASE, self.auth_type, self.mode), batch_refund_id),
            headers=self.headers,
        )
        batch_refund_status_response_schema = RefundStatusByIdentifierResponseSchema()
        return batch_refund_status_response_schema.load(api_response.json()['data'])

    @Decorators.auth_handler
    def get_refund_status_by_identifier(
        self, identifier_type: RefundStatusIdentifierType, identifier_value: str
    ) -> RefundStatusByIdentifierResponse:
        """Get batch refund status."""
        api_response = self.session.get(
            "{}/{}/{}".format(
                get_url_path(API.REFUNDS_BASE, self.auth_type, self.mode), identifier_type, identifier_value
            ),
            headers=self.headers,
        )
        batch_refund_status_response_schema = RefundStatusByIdentifierResponseSchema()
        return batch_refund_status_response_schema.load(api_response.json()['data'])

    @Decorators.auth_handler
    def get_refund_status(self, refund_id: str) -> RefundResponseItem:
        """Get individual refund status."""
        api_response = self.session.get(
            "{}/{}".format(get_url_path(API.REFUNDS_BASE, self.auth_type, self.mode), refund_id),
            headers=self.headers,
        )
        refund_response_item_schema = RefundResponseItemSchema()
        return refund_response_item_schema.load(api_response.json()['data'])

`Decorators` ¶

Decorators.

Source code in setu/deeplink.py

class Decorators:
    """Decorators."""

    @staticmethod
    def auth_handler(decorated: Callable):
        """Refresh token."""

        def wrapper(deeplink, *args, **kwargs):
            try:
                if deeplink.auth_type == AUTH_TYPE_JWT:
                    deeplink.regenerate_token()
                return decorated(deeplink, *args, **kwargs)
            except requests.exceptions.HTTPError:
                deeplink.regenerate_token()
                return decorated(deeplink, *args, **kwargs)

        return wrapper

`auth_handler(decorated)` `staticmethod` ¶

Refresh token.

Source code in setu/deeplink.py

@staticmethod
def auth_handler(decorated: Callable):
    """Refresh token."""

    def wrapper(deeplink, *args, **kwargs):
        try:
            if deeplink.auth_type == AUTH_TYPE_JWT:
                deeplink.regenerate_token()
            return decorated(deeplink, *args, **kwargs)
        except requests.exceptions.HTTPError:
            deeplink.regenerate_token()
            return decorated(deeplink, *args, **kwargs)

    return wrapper

`init(self, scheme_id, secret, product_instance_id, auth_type='JWT', mode='SANDBOX')` `special` ¶

Constructor for the Deeplink class.

Parameters:

Name	Type	Description	Default
`scheme_id`	`str`	description	required
`secret`	`str`	description	required
`product_instance_id`	`str`	description	required
`auth_type`	`Literal["JWT", "OAUTH"]`	description. Defaults to "JWT".	`'JWT'`
`mode`	`Literal["SANDBOX", "PRODUCTION"]`	description. Defaults to "SANDBOX".	`'SANDBOX'`

Source code in setu/deeplink.py

def __init__(
    self,
    scheme_id: str,
    secret: str,
    product_instance_id: str,
    auth_type: AuthType = "JWT",
    mode: Mode = "SANDBOX",
):
    """Constructor for the Deeplink class.

    Args:
        scheme_id (str): _description_
        secret (str): _description_
        product_instance_id (str): _description_
        auth_type (Literal["JWT", "OAUTH"], optional): _description_. Defaults to "JWT".
        mode (Literal["SANDBOX", "PRODUCTION"], optional): _description_. Defaults to "SANDBOX".
    """
    self.scheme_id = scheme_id
    self.secret = secret
    self.mode = mode
    self.auth_type = auth_type

    self.headers = {
        "X-Setu-Product-Instance-ID": product_instance_id,
        "Content-Type": "application/json",
    }

    self.session = requests.Session()
    self.session.hooks = {"response": lambda r, *args, **kwargs: self.exception_handler(r)}

`exception_handler(r)` `staticmethod` ¶

Exception handler.

Source code in setu/deeplink.py

@staticmethod
def exception_handler(r: Response):
    """Exception handler."""
    try:
        r.raise_for_status()
    except requests.exceptions.RequestException as e:
        if isinstance(e, requests.exceptions.HTTPError) and r.reason == "Forbidden":
            raise e
        else:
            error_response_schema = SetuErrorResponseSchema()
            raise SetuAPIException(error_response_schema.load(r.json()))

`regenerate_token(self)` ¶

Re-generate token.

Source code in setu/deeplink.py

def regenerate_token(self):
    """Re-generate token."""
    if self.auth_type == AUTH_TYPE_JWT:
        authorization = generate_jwt_token(self.scheme_id, self.secret)
    else:
        authorization = generate_oauth_token(self.scheme_id, self.secret, self.mode)
    self.headers["Authorization"] = authorization

`endpoint` ¶

Endpoint module.

`get_url_path(endpoint, auth_type='JWT', mode='SANDBOX')` ¶

Get URL for API.

Parameters:

Name	Type	Description	Default
`endpoint`	`API`	description	required
`auth_type`	`AuthType`	description. Defaults to "JWT".	`'JWT'`
`mode`	`Mode`	description. Defaults to "SANDBOX".	`'SANDBOX'`

Returns:

Type	Description
`str`	description

Source code in setu/endpoint.py

def get_url_path(
    endpoint: API,
    auth_type: AuthType = "JWT",
    mode: Mode = "SANDBOX",
) -> str:
    """Get URL for API.

    Args:
        endpoint (API): _description_
        auth_type (AuthType, optional): _description_. Defaults to "JWT".
        mode (Mode, optional): _description_. Defaults to "SANDBOX".

    Returns:
        str: _description_
    """
    return "{base_url}{api_version}{path}".format(
        base_url=PRODUCTION_BASE if mode == MODE_PRODUCTION else SANDBOX_BASE,
        api_version="/v2" if auth_type == AUTH_TYPE_OAUTH else "",
        path=endpoint.value,
    )