From 760991fd6dbb211849c9123b6578b85c2c49f4c2 Mon Sep 17 00:00:00 2001 From: ChemicalLuck Date: Wed, 1 May 2024 10:06:11 +0100 Subject: [PATCH] Initial rewrite --- .gitignore | 10 +- LICENSE | 2 +- justfile | 13 + pyproject.toml | 3 + recharge/__init__.py | 69 ++--- recharge/api/__init__.py | 104 ++++++++ recharge/api/addresses.py | 183 ++++++++++++++ recharge/api/async_batches.py | 95 +++++++ recharge/api/charges.py | 169 +++++++++++++ recharge/api/checkouts.py | 194 ++++++++++++++ recharge/api/customers.py | 132 ++++++++++ recharge/api/discounts.py | 141 +++++++++++ recharge/api/metafields.py | 146 +++++++++++ recharge/api/notifications.py | 35 +++ recharge/api/onetimes.py | 96 +++++++ recharge/api/orders.py | 194 ++++++++++++++ recharge/api/products.py | 141 +++++++++++ recharge/api/shop.py | 28 ++ recharge/api/subscriptions.py | 250 ++++++++++++++++++ recharge/api/tokens.py | 61 +++++ recharge/api/webhooks.py | 123 +++++++++ recharge/resources.py | 464 ---------------------------------- requirements.txt | 11 + setup.cfg | 28 ++ setup.py | 39 --- 25 files changed, 2189 insertions(+), 542 deletions(-) create mode 100644 justfile create mode 100644 pyproject.toml create mode 100644 recharge/api/__init__.py create mode 100644 recharge/api/addresses.py create mode 100644 recharge/api/async_batches.py create mode 100644 recharge/api/charges.py create mode 100644 recharge/api/checkouts.py create mode 100644 recharge/api/customers.py create mode 100644 recharge/api/discounts.py create mode 100644 recharge/api/metafields.py create mode 100644 recharge/api/notifications.py create mode 100644 recharge/api/onetimes.py create mode 100644 recharge/api/orders.py create mode 100644 recharge/api/products.py create mode 100644 recharge/api/shop.py create mode 100644 recharge/api/subscriptions.py create mode 100644 recharge/api/tokens.py create mode 100644 recharge/api/webhooks.py delete mode 100644 recharge/resources.py create mode 100644 requirements.txt create mode 100644 setup.cfg delete mode 100644 setup.py diff --git a/.gitignore b/.gitignore index 94cb12a..48571a9 100644 --- a/.gitignore +++ b/.gitignore @@ -1,8 +1,4 @@ -*.pyc +/.ruff_cache __pycache__ -.vscode -venv -.idea -dist/* -recharge_api.egg-info -recharge-api.toml \ No newline at end of file +/.venv +/*.egg-info diff --git a/LICENSE b/LICENSE index 664a71c..f935922 100644 --- a/LICENSE +++ b/LICENSE @@ -16,4 +16,4 @@ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -THE SOFTWARE. \ No newline at end of file +THE SOFTWARE. diff --git a/justfile b/justfile new file mode 100644 index 0000000..95cdf3d --- /dev/null +++ b/justfile @@ -0,0 +1,13 @@ +alias l := lock +alias i := install +alias u := uninstall +alias c := clean + +lock: + uv pip freeze | uv pip compile - -o requirements.txt +install: + uv pip install -e . +uninstall: + uv pip uninstall qrgenerator +clean: + rm -rf *.egg-info .ruff_cache diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..e9a14aa --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,3 @@ +[build-system] +requires = ["setuptools >= 48"] +build-backend = "setuptools.build_meta" diff --git a/recharge/__init__.py b/recharge/__init__.py index 04af9e6..8e4597a 100644 --- a/recharge/__init__.py +++ b/recharge/__init__.py @@ -1,39 +1,46 @@ -from .resources import ( - RechargeAddress, - RechargeCharge, - RechargeCheckout, - RechargeCustomer, - RechargeOrder, - RechargeSubscription, - RechargeOnetime, - RechargeDiscount, - RechargeWebhook, - RechargeMetafield, - RechargeShop, - RechargeProduct -) +from recharge.api.notifications import NotificationResource +from recharge.api.addresses import AddressResource +from recharge.api.charges import ChargeResource +from recharge.api.checkouts import CheckoutResource +from recharge.api.customers import CustomerResource +from recharge.api.orders import OrderResource +from recharge.api.subscriptions import SubscriptionResource +from recharge.api.onetimes import OnetimeResource +from recharge.api.discounts import DiscountResource +from recharge.api.webhooks import WebhookResource +from recharge.api.metafields import MetafieldResource +from recharge.api.shop import ShopResource +from recharge.api.products import ProductResource +from recharge.api.async_batches import AsyncBatchResource +from recharge.api.tokens import TokenResource class RechargeAPI(object): - - def __init__(self, access_token=None, log_debug=False): + def __init__(self, access_token=None, debug=False): self.access_token = access_token - self.log_debug = log_debug + self.debug = debug kwargs = { - 'access_token': access_token, - 'log_debug': log_debug, + "access_token": access_token, + "log_debug": debug, } - self.Address = RechargeAddress(**kwargs) - self.Charge = RechargeCharge(**kwargs) - self.Checkout = RechargeCheckout(**kwargs) - self.Customer = RechargeCustomer(**kwargs) - self.Order = RechargeOrder(**kwargs) - self.Subscription = RechargeSubscription(**kwargs) - self.Onetime = RechargeOnetime(**kwargs) - self.Discount = RechargeDiscount(**kwargs) - self.Webhook = RechargeWebhook(**kwargs) - self.Metafield = RechargeMetafield(**kwargs) - self.Shop = RechargeShop(**kwargs) - self.Product = RechargeProduct(**kwargs) + self.Token = TokenResource(**kwargs) + self.scopes = self.Token.get()["scopes"] + + kwargs["scopes"] = self.scopes + + self.Address = AddressResource(**kwargs) + self.Charge = ChargeResource(**kwargs) + self.Checkout = CheckoutResource(**kwargs) + self.Customer = CustomerResource(**kwargs) + self.Order = OrderResource(**kwargs) + self.Subscription = SubscriptionResource(**kwargs) + self.Onetime = OnetimeResource(**kwargs) + self.Discount = DiscountResource(**kwargs) + self.Webhook = WebhookResource(**kwargs) + self.Metafield = MetafieldResource(**kwargs) + self.Shop = ShopResource(**kwargs) + self.Product = ProductResource(**kwargs) + self.AsyncBatch = AsyncBatchResource(**kwargs) + self.Notification = NotificationResource(**kwargs) diff --git a/recharge/api/__init__.py b/recharge/api/__init__.py new file mode 100644 index 0000000..754ec12 --- /dev/null +++ b/recharge/api/__init__.py @@ -0,0 +1,104 @@ +import logging +import time +from typing import Any, Mapping + +import requests + +from recharge.api.tokens import TokenScope + +log = logging.getLogger(__name__) + + +class RechargeResource(object): + """ + Resource from the Recharge API. This class handles + logging, sending requests, parsing JSON, and rate + limiting. + + Refer to the API docs to see the expected responses. + https://developer.rechargepayments.com/ + """ + + base_url = "https://api.rechargeapps.com" + object_list_key = None + + def __init__( + self, access_token=None, debug=False, scopes: list[TokenScope] | None = None + ): + self.debug = debug + self.headers = { + "Accept": "application/json", + "Content-Type": "application/json", + "X-Recharge-Access-Token": access_token, + } + self.scopes = scopes + self.allowed_endpoints = [] + + def check_scopes(self, endpoint: str, scopes: list[TokenScope]): + if endpoint in self.allowed_endpoints: + return + + if not self.scopes: + raise ValueError("No scopes found for token.") + + missing_scopes = [] + + for scope in scopes: + if scope not in self.scopes: + missing_scopes.append(scope) + + if missing_scopes: + raise ValueError(f"Endpoint {endpoint} missing scopes: {missing_scopes}") + else: + self.allowed_endpoints.append(endpoint) + + def log(self, url, response): + if self.debug: + log.info(url) + log.info(response.headers["X-Recharge-Limit"]) + + @property + def url(self) -> str: + return f"{self.base_url}/{self.object_list_key}" + + def http_delete(self, url: str, body: Mapping[str, Any] | None = None): + response = requests.delete(url, headers=self.headers, json=body) + log.info(url) + log.info(response.headers["X-Recharge-Limit"]) + if response.status_code == 429: + return self.http_delete(url) + return response + + def http_get(self, url: str, query: Mapping[str, Any] | None = None): + response = requests.get(url, params=query, headers=self.headers) + self.log(url, response) + if response.status_code == 429: + time.sleep(1) + return self.http_get(url) + return response.json() + + def http_put( + self, + url: str, + body: Mapping[str, Any] | None = None, + query: Mapping[str, Any] | None = None, + ): + response = requests.put(url, json=body, params=query, headers=self.headers) + self.log(url, response) + if response.status_code == 429: + time.sleep(1) + return self.http_put(url, body) + return response.json() + + def http_post( + self, + url: str, + body: Mapping[str, Any] | None = None, + query: Mapping[str, Any] | None = None, + ): + response = requests.post(url, json=body, params=query, headers=self.headers) + self.log(url, response) + if response.status_code == 429: + time.sleep(1) + return self.http_post(url, body) + return response.json() diff --git a/recharge/api/addresses.py b/recharge/api/addresses.py new file mode 100644 index 0000000..1cf1f4e --- /dev/null +++ b/recharge/api/addresses.py @@ -0,0 +1,183 @@ +from typing import Required, TypedDict + +from recharge.api import RechargeResource +from recharge.api.tokens import TokenScope + + +class AddressNoteAttributes(TypedDict): + name: str + value: str + + +class AddressShippingLinesOverride(TypedDict): + code: str + price: str + title: str + + +class AddressCreateBody(TypedDict): + address1: Required[str] + address2: Required[str] + cart_note: str + city: Required[str] + company: str + country: Required[str] + first_name: Required[str] + last_name: Required[str] + note_attributes: list[AddressNoteAttributes] + phone: Required[str] + presentment_currency: str + province: Required[str] + shipping_lines_override: list[AddressShippingLinesOverride] + zip: Required[str] + + +class AddressUpdateBody(TypedDict): + address1: str + address2: str + cart_note: str + city: str + company: str + country: str + first_name: str + last_name: str + note_attributes: list[AddressNoteAttributes] + phone: str + province: str + shipping_lines_override: list[AddressShippingLinesOverride] + zip: str + + +class AddressListQuery(TypedDict): + created_at_max: str + created_at_min: str + customer_id: str + discount_code: str + discount_id: str + ids: str + limit: int + page: int + updated_at_max: str + updated_at_min: str + + +class AddressCountQuery(TypedDict): + created_at_max: str + created_at_min: str + discount_code: str + discount_id: str + updated_at_max: str + updated_at_min: str + + +class AddressValidateBody(TypedDict): + address1: str + city: str + state: str + zipcode: str + + +class AddressApplyDiscountCode(TypedDict): + discount_code: str + + +class AddressApplyDiscountId(TypedDict): + discount_id: str + + +class AddressResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/addresses + """ + + object_list_key = "addresses" + + def create(self, customer_id, body: AddressCreateBody): + """Create an address for the customer. + https://developer.rechargepayments.com/2021-01/addresses/create_address + """ + required_scopes: list[TokenScope] = ["write_customers"] + self.check_scopes("POST /customers/:customer_id/addresses", required_scopes) + + url = f"{self.base_url}/customers/{customer_id}/{self.object_list_key}" + return self.http_post(url, body) + + def get(self, address_id: str): + """Get an address by ID. + https://developer.rechargepayments.com/2021-01/addresses/retrieve_address + """ + required_scopes: list[TokenScope] = ["read_customers"] + self.check_scopes(f"GET {self.object_list_key}/:address_id", required_scopes) + + return self.http_get(f"{self.url}/{address_id}") + + def update(self, address_id, body: AddressUpdateBody | None = None): + """Update an address by ID. + https://developer.rechargepayments.com/2021-01/addresses/update_address + """ + required_scopes: list[TokenScope] = ["write_customers"] + self.check_scopes(f"PUT {self.object_list_key}/:id", required_scopes) + + return self.http_put(f"{self.url}/{address_id}", body) + + def delete(self, address_id): + """Delete an address by ID. + https://developer.rechargepayments.com/2021-01/addresses/delete_address + """ + required_scopes: list[TokenScope] = ["write_customers"] + self.check_scopes(f"DELETE {self.object_list_key}/:address_id", required_scopes) + + return self.http_delete(f"{self.url}/{address_id}") + + def list(self, customer_id, query: AddressListQuery | None = None): + """List all addresses for a customer. + https://developer.rechargepayments.com/2021-01/addresses/list_addresses + """ + required_scopes: list[TokenScope] = ["read_customers"] + self.check_scopes( + f"GET /customers/:customer_id/{self.object_list_key}", required_scopes + ) + + return self.http_get( + f"{self.base_url}/customers/{customer_id}/{self.object_list_key}", query + ) + + def count(self, query: AddressCountQuery | None = None): + """Retrieve the count of addresses. + https://developer.rechargepayments.com/2021-01/addresses/count_addresses + """ + required_scopes: list[TokenScope] = ["read_customers"] + self.check_scopes("GET /addresses/count", required_scopes) + + return self.http_get(f"{self.url}/count", query) + + def validate(self, body: AddressValidateBody): + """Validate an address. + https://developer.rechargepayments.com/2021-01/addresses/validate_address + """ + + return self.http_post(f"{self.url}/validate_address", body) + + def apply_discount( + self, + address_id, + body: AddressApplyDiscountCode | AddressApplyDiscountId, + ): + """Apply a discount code to an address. + https://developer.rechargepayments.com/2021-01/discounts/discounts_apply_address + """ + required_scopes: list[TokenScope] = ["write_discounts"] + self.check_scopes("POST /addresses/:address_id/apply_discount", required_scopes) + + return self.http_post(f"{self.url}/{address_id}/apply_discount", body) + + def remove_discount(self, address_id: str): + """Remove a discount from an address. + https://developer.rechargepayments.com/2021-01/discounts/discounts_remove_from_address_or_charge + """ + required_scopes: list[TokenScope] = ["write_discounts"] + self.check_scopes( + "POST /addresses/:address_id/remove_discount", required_scopes + ) + + return self.http_post(f"{self.url}/{address_id}/remove_discount") diff --git a/recharge/api/async_batches.py b/recharge/api/async_batches.py new file mode 100644 index 0000000..5a89ad0 --- /dev/null +++ b/recharge/api/async_batches.py @@ -0,0 +1,95 @@ +from recharge.api import RechargeResource +from typing import TypedDict, Literal + +from recharge.api.tokens import TokenScope + + +class AsyncBatchCreateBody(TypedDict): + batch_type: Literal[ + "address_discount_apply", + "address_discount_remove", + "change_next_charge_date", + "discount_create", + "discount_delete", + "discount_update", + "product_create", + "product_update", + "product_delete", + "onetime_create", + "onetime_delete", + "bulk_subscriptions_create", + "bulk_subscriptions_update", + "bulk_subscriptions_delete", + "subscription_cancel", + ] + + +class AsyncBatchResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/async_batch_endpoints + """ + + object_list_key = "async_batches" + + def create(self, data): + """Create an async batch. + https://developer.rechargepayments.com/2021-01/async_batch_endpoints + """ + required_scopes: list[TokenScope] = ["write_batches"] + self.check_scopes("POST /async_batches", required_scopes) + + return self.http_post(self.url, data) + + def create_task(self, batch_id, data): + """Create a task for an async batch. + https://developer.rechargepayments.com/2021-01/async_batch_endpoints + """ + required_scopes: list[TokenScope] = ["write_batches"] + self.check_scopes( + f"POST /{self.object_list_key}/:batch_id/tasks", required_scopes + ) + + return self.http_post(f"{self.url}/{batch_id}/tasks", data) + + def get(self, batch_id): + """Get an async batch. + https://developer.rechargepayments.com/2021-01/async_batch_endpoints + """ + required_scopes: list[TokenScope] = ["read_batches"] + self.check_scopes("GET /async_batches/:batch_id", required_scopes) + + return self.http_get(f"{self.url}/{batch_id}") + + def list(self): + """List async batches. + https://developer.rechargepayments.com/2021-01/async_batch_endpoints + """ + required_scopes: list[TokenScope] = ["read_batches"] + self.check_scopes("GET /async_batches", required_scopes) + + return self.http_get(self.url) + + def list_tasks( + self, + batch_id, + ): + """List tasks for an async batch. + https://developer.rechargepayments.com/2021-01/async_batch_endpoints + """ + required_scopes: list[TokenScope] = ["read_batches"] + self.check_scopes( + f"GET /{self.object_list_key}/:batch_id/tasks", required_scopes + ) + + return self.http_get(f"{self.url}/{batch_id}/tasks") + + def process(self, batch_id): + """Process an async batch. + https://developer.rechargepayments.com/2021-01/async_batch_endpoints + """ + required_scopes: list[TokenScope] = ["write_batches"] + self.check_scopes( + f"POST /{self.object_list_key}/:batch_id/process", required_scopes + ) + + return self.http_post(f"{self.url}/{batch_id}/process", None) diff --git a/recharge/api/charges.py b/recharge/api/charges.py new file mode 100644 index 0000000..2421186 --- /dev/null +++ b/recharge/api/charges.py @@ -0,0 +1,169 @@ +from typing import Literal, Required, TypedDict + +from recharge.api import RechargeResource +from recharge.api.tokens import TokenScope + +type ChargeStatus = Literal[ + "SUCCESS", "QUEUED", "ERROR", "REFUNDED", "PARTIALLY_REFUNDED", "SKIPPED" +] + + +class ChargeListQuery(TypedDict): + address_id: str + created_at_max: str + created_at_min: str + customer_id: str + date: str + date_max: str + date_min: str + discount_code: str + discount_id: str + ids: str + limit: int + page: int + shopify_order_id: str + status: ChargeStatus + subscription_id: str + updated_at_max: str + updated_at_min: str + + +class ChargeCountQuery(TypedDict): + address_id: str + customer_id: str + date: str + date_max: str + date_min: str + discount_id: str + shopify_order_id: str + status: ChargeStatus + subscription_id: str + + +class ChargeChangeNextChargeDateBody(TypedDict, total=True): + next_charge_date: str + + +class ChargeSkipSubscriptionId(TypedDict, total=True): + subscription_id: str + + +class ChargeSkupSubscriptionIds(TypedDict, total=True): + subscription_ids: list[str] + + +type ChargeSkipBody = ChargeSkipSubscriptionId | ChargeSkupSubscriptionIds + + +class ChargeRefundBody(TypedDict): + amount: Required[str] + full_refund: bool + + +class ChargeResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/charges + """ + + object_list_key = "charges" + + def get(self, charge_id: str): + """Get a charge by id. + https://developer.rechargepayments.com/2021-01/charges/charge_retrieve + """ + required_scopes: list[TokenScope] = ["read_orders"] + self.check_scopes(f"GET {self.object_list_key}/:charge_id", required_scopes) + + return self.http_get(f"{self.url}/{charge_id}") + + def list(self, query: ChargeListQuery | None = None): + """List charges. + https://developer.rechargepayments.com/2021-01/charges/charge_list + """ + required_scopes: list[TokenScope] = ["read_orders"] + self.check_scopes(f"GET {self.object_list_key}", required_scopes) + + return self.http_get(self.url, query) + + def count(self, query: ChargeCountQuery | None = None): + """Count charges. + https://developer.rechargepayments.com/2021-01/charges/charge_count + """ + required_scopes: list[TokenScope] = ["read_orders"] + self.check_scopes(f"GET {self.object_list_key}/count", required_scopes) + + return self.http_get(f"{self.url}/count", query) + + def change_next_charge_date( + self, charge_id: str, body: ChargeChangeNextChargeDateBody + ): + """Change the date of a queued charge. + https://developer.rechargepayments.com/2021-01/charges/charge_change_next_date + """ + required_scopes: list[TokenScope] = ["write_orders"] + self.check_scopes( + f"POST {self.object_list_key}/:charge_id/change_next_charge_date", + required_scopes, + ) + + return self.http_put(f"{self.url}/{charge_id}/change_next_charge_date", body) + + def skip(self, charge_id: str, body: ChargeSkipBody): + """Skip a charge. + https://developer.rechargepayments.com/2021-01/charges/charge_skip + """ + required_scopes: list[TokenScope] = ["write_orders"] + self.check_scopes( + f"POST {self.object_list_key}/:charge_id/skip", required_scopes + ) + + return self.http_post(f"{self.url}/{charge_id}/skip", body) + + def unskip(self, charge_id: str, body: ChargeSkipBody): + """Unskip a charge. + https://developer.rechargepayments.com/2021-01/charges/charge_unskip + """ + required_scopes: list[TokenScope] = ["write_orders"] + self.check_scopes( + f"POST {self.object_list_key}/:charge_id/unskip", required_scopes + ) + + return self.http_post(f"{self.url}/{charge_id}/unskip", body) + + def refund(self, charge_id: str, body: ChargeRefundBody): + """Refund a charge. + https://developer.rechargepayments.com/2021-01/charges/charge_refund + """ + required_scopes: list[TokenScope] = ["write_orders", "write_payments"] + self.check_scopes( + f"POST {self.object_list_key}/:charge_id/refund", required_scopes + ) + + return self.http_post(f"{self.url}/{charge_id}/refund", body) + + def process(self, charge_id: str): + """Process a charge. + https://developer.rechargepayments.com/2021-01/charges/charge_process + """ + required_scopes: list[TokenScope] = ["write_payments"] + self.check_scopes( + f"POST {self.object_list_key}/:charge_id/process", required_scopes + ) + + return self.http_post(f"{self.url}/{charge_id}/process", None) + + def capture(self, charge_id: str): + """Capture a charge. + https://developer.rechargepayments.com/2021-01/charges/charge_capture + """ + required_scopes: list[TokenScope] = [ + "write_orders", + "write_payments", + "write_subscriptions", + "write_customers", + ] + self.check_scopes( + f"POST {self.object_list_key}/:charge_id/capture", required_scopes + ) + + return self.http_post(f"{self.url}/{charge_id}/capture_payment", None) diff --git a/recharge/api/checkouts.py b/recharge/api/checkouts.py new file mode 100644 index 0000000..26b0030 --- /dev/null +++ b/recharge/api/checkouts.py @@ -0,0 +1,194 @@ +from typing import Literal, TypedDict, Required + +from recharge.api import RechargeResource +from recharge.api.tokens import TokenScope + + +class CheckoutUtmParams(TypedDict): + utm_campaign: str + utm_content: str + utm_data_source: str + utm_source: str + utm_medium: str + utm_term: str + utm_timestamp: str + + +class CheckoutAnalyticsData(TypedDict): + utm_params: list[CheckoutUtmParams] + + +class CheckoutBillingAddress(TypedDict): + address1: str + address2: str + city: str + company: str + country: str + first_name: str + last_name: str + phone: str + province: str + zip: str + + +type CheckoutExternalCheckoutSource = Literal["big_commerce", "headless", "shopify"] + +type CheckoutOrderIntervalUnit = Literal["day", "week", "month"] + + +class CheckoutLineItemProperty(TypedDict): + name: str + value: str + + +type CheckoutLineItemType = Literal["SUBSCRIPTION", "ONETIME"] + + +class CheckoutLineItem(TypedDict): + charge_interval_frequency: int + cutoff_day_of_month: int + cutoff_day_of_week: int + expire_after_specific_number_of_charges: int + first_recurring_charge_delay: int + fulfillment_service: str + grams: int + image: str + order_day_of_month: int + order_day_of_week: int + order_interval_frequency: int + order_interval_unit: CheckoutOrderIntervalUnit + original_price: str + price: str + product_id: int + product_type: str + properties: list[CheckoutLineItemProperty] + quantity: int + requires_shipping: bool + sku: str + tax_code: str + taxable: bool + title: str + type: CheckoutLineItemType + variant_id: int + variant_title: str + + +class CheckoutNoteAttribute(TypedDict): + name: str + value: str + + +class CheckoutShippingAddress(TypedDict): + address1: str + address2: str + city: str + company: str + country: str + first_name: str + last_name: str + phone: str + province: str + zip: str + + +class CheckoutCreateBody(TypedDict): + analytics_data: CheckoutAnalyticsData + billing_address: CheckoutBillingAddress + buyer_accepts_marketing: bool + currency: str + discount_code: str + email: str + external_checkout_id: str + external_checkout_source: CheckoutExternalCheckoutSource + external_checkout_customer_id: str + line_items: Required[list[CheckoutLineItem]] + note: str + note_attributes: list[CheckoutNoteAttribute] + phone: str + shipping_address: CheckoutShippingAddress + + +class CheckoutUpdateBody(TypedDict): + analytics_data: CheckoutAnalyticsData + billing_address: CheckoutBillingAddress + buyer_accepts_marketing: bool + currency: str + discount_code: str + email: str + external_checkout_id: str + external_checkout_source: CheckoutExternalCheckoutSource + external_checkout_customer_id: str + line_items: list[CheckoutLineItem] + note: str + note_attributes: list[CheckoutNoteAttribute] + partial_shipping: bool + phone: str + shipping_address: CheckoutShippingAddress + + +type CheckoutPaymentProcessor = Literal["stripe", "braintree", "mollie", "authorize"] + +type CheckoutPaymentType = Literal["CREDIT_CARD", "PAYPAL", "APPLE_PAY", "GOOGLE_PAY"] + + +class CheckoutProcessBody(TypedDict): + payment_processor: Required[CheckoutPaymentProcessor] + payment_token: Required[str] + payment_type: CheckoutPaymentType + + +class CheckoutResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/checkouts + """ + + object_list_key = "checkouts" + + def create(self, body: CheckoutCreateBody): + """Create a new checkout. + https://developer.rechargepayments.com/2021-01/checkouts/checkout_create + """ + required_scopes: list[TokenScope] = ["write_checkouts"] + self.check_scopes(f"POST /{self.object_list_key}", required_scopes) + + return self.http_post(self.url, body) + + def get(self, checkout_id: str): + """Get a checkout by ID. + https://developer.rechargepayments.com/2021-01/checkouts/checkout_retrieve + """ + required_scopes: list[TokenScope] = ["read_checkouts"] + self.check_scopes(f"GET /{self.object_list_key}/:checkout_id", required_scopes) + + return self.http_get(f"{self.url}/{checkout_id}") + + def update(self, checkout_id: str, body: CheckoutUpdateBody): + """Update a checkout. + https://developer.rechargepayments.com/2021-01/checkouts/checkout_update + """ + required_scopes: list[TokenScope] = ["write_checkouts"] + self.check_scopes(f"PUT /{self.object_list_key}/:checkout_id", required_scopes) + + return self.http_put(f"{self.url}/{checkout_id}", body) + + def get_shipping(self, checkout_id: str): + """Retrieve shipping rates for a checkout + https://developer.rechargepayments.com/2021-01/checkouts/checkout_retrieve_shipping_address + """ + required_scopes: list[TokenScope] = ["read_checkouts"] + self.check_scopes( + f"GET /{self.object_list_key}/:checkout_id/shipping_rates", required_scopes + ) + + return self.http_get(f"{self.url}/{checkout_id}/shipping_rates") + + def process(self, checkout_id: str, body: CheckoutProcessBody): + """Process (charge) a checkout. + https://developer.rechargepayments.com/2021-01/checkout/checkout_process + """ + required_scopes: list[TokenScope] = ["write_checkouts"] + self.check_scopes( + f"POST /{self.object_list_key}/:checkout_id/charge", required_scopes + ) + + return self.http_post(f"{self.url}/{checkout_id}/charge", body) diff --git a/recharge/api/customers.py b/recharge/api/customers.py new file mode 100644 index 0000000..e3d9326 --- /dev/null +++ b/recharge/api/customers.py @@ -0,0 +1,132 @@ +from recharge.api import RechargeResource +from recharge.api.tokens import TokenScope + +from typing import TypedDict, Required, Literal + + +class CustomerCreateBody(TypedDict): + accepts_marketing: bool + billing_address1: Required[str] + billing_address2: str + billing_city: Required[str] + billing_company: str + billing_country: Required[str] + billing_first_name: Required[str] + billing_last_name: Required[str] + billing_phone: str + billing_province: Required[str] + billing_zip: Required[str] + email: Required[str] + first_name: Required[str] + last_name: Required[str] + phone: str + processor_type: str + shopify_customer_id: str + + +class CustomerUpdateBody(TypedDict): + accepts_marketing: bool + billing_address1: str + billing_address2: str + billing_city: str + billing_company: str + billing_country: str + billing_first_name: str + billing_last_name: str + billing_phone: str + billing_province: str + billing_zip: str + email: str + first_name: str + last_name: str + phone: str + shopify_customer_id: str + + +type CustomerStatus = Literal["ACTIVE", "INACTIVE"] + + +class CustomerListQuery(TypedDict): + email: str + created_at_max: str + created_at_min: str + hash: str + ids: str + limit: str + page: str + shopify_customer_id: str + status: CustomerStatus + updated_at_max: str + updated_at_min: str + + +class CustomerCountQuery(TypedDict): + created_at_max: str + created_at_min: str + status: CustomerStatus + updated_at_max: str + updated_at_min: str + + +class CustomerResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/customers + """ + + object_list_key = "customers" + + def create(self, body: CustomerCreateBody): + """Create a customer. + https://developer.rechargepayments.com/2021-01/customers/customers_create + """ + required_scopes: list[TokenScope] = ["write_customers", "write_payments"] + self.check_scopes(f"POST /{self.object_list_key}", required_scopes) + + return self.http_post(self.url, body) + + def get(self, customer_id: str): + """Get a customer by ID. + https://developer.rechargepayments.com/2021-01/customers/customers_retrieve + """ + required_scopes: list[TokenScope] = ["read_customers"] + self.check_scopes(f"GET /{self.object_list_key}/:customer_id", required_scopes) + + return self.http_get(f"{self.url}/{customer_id}") + + def update(self, customer_id: str, body: CustomerUpdateBody): + """Update a customer. + https://developer.rechargepayments.com/2021-01/customers/customers_update + """ + required_scopes: list[TokenScope] = ["write_customers"] + self.check_scopes(f"PUT /{self.object_list_key}/:customer_id", required_scopes) + + return self.http_put(f"{self.url}/{customer_id}", body) + + def delete(self, customer_id: str): + """Delete a customer. + https://developer.rechargepayments.com/2021-01/customers/customers_delete + """ + required_scopes: list[TokenScope] = ["write_customers"] + self.check_scopes( + f"DELETE /{self.object_list_key}/:customer_id", required_scopes + ) + + return self.http_delete(f"{self.url}/{customer_id}") + + def list(self, query: CustomerListQuery | None = None): + """List customers. + https://developer.rechargepayments.com/2021-01/customers/customers_list + """ + required_scopes: list[TokenScope] = ["read_customers"] + self.check_scopes(f"GET /{self.object_list_key}", required_scopes) + + return self.http_get(self.url, query) + + def count(self, query: CustomerCountQuery | None = None): + """Retrieve a count of customers. + https://developer.rechargepayments.com/2021-01/customers/customers_count + """ + required_scopes: list[TokenScope] = ["read_customers"] + self.check_scopes(f"GET /{self.object_list_key}/count", required_scopes) + + return self.http_get(f"{self.url}/count", query) diff --git a/recharge/api/discounts.py b/recharge/api/discounts.py new file mode 100644 index 0000000..1803bce --- /dev/null +++ b/recharge/api/discounts.py @@ -0,0 +1,141 @@ +from typing import Literal, Required, TypedDict + +from recharge.api import RechargeResource +from recharge.api.tokens import TokenScope + +type DiscountProductType = Literal["ALL", "ONETIME", "SUBSCRIPTION"] + +type DiscountAppliesToResource = Literal["shopify_product", "shopify_collection_id"] + + +class DiscountChannelSettingsValue: + can_apply: bool + + +class DiscountChannelSettings(TypedDict): + api: DiscountChannelSettingsValue + checkout_page: DiscountChannelSettingsValue + customer_portal: DiscountChannelSettingsValue + merchant_portal: DiscountChannelSettingsValue + + +type DiscountType = Literal["percentage", "fixed_amount"] + +type DiscountFirstTimeCustomerRestriction = Literal[ + "null", "customer_must_not_exist_in_recharge" +] + +type DiscountStatus = Literal["enabled", "disabled", "fully_disabled"] + + +class DiscountCreateBody(TypedDict): + applies_to_id: int + applies_to_product_type: DiscountProductType + applies_to_resource: DiscountAppliesToResource + channel_settings: DiscountChannelSettings + code: Required[str] + discount_type: DiscountType + duration: str + duration_usage_limit: int + ends_at: str + first_time_customer_restriction: DiscountFirstTimeCustomerRestriction + once_per_customer: bool + prerequisite_subtotal_min: int + starts_at: str + status: DiscountStatus + usage_limit: int + value: str + + +class DiscountUpdateBody(TypedDict): + channel_settings: DiscountChannelSettings + ends_at: str + starts_at: str + status: DiscountStatus + usage_limit: int + + +class DiscountListQuery(TypedDict): + created_at_max: str + created_at_min: str + discount_code: str + discount_type: DiscountType + ids: str + limit: str + page: str + status: DiscountStatus + updated_at_max: str + updated_at_min: str + + +class DiscountCountQuery(TypedDict): + created_at_max: str + created_at_min: str + discount_type: DiscountType + status: DiscountStatus + updated_at_max: str + updated_at_min: str + + +class DiscountResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/discounts + """ + + object_list_key = "discounts" + + def create(self, body: DiscountCreateBody): + """Create a discount. + https://developer.rechargepayments.com/2021-01/discounts/discounts_create + """ + required_scopes: list[TokenScope] = ["write_discounts"] + self.check_scopes(f"POST /{self.object_list_key}", required_scopes) + + return self.http_post(self.url, body) + + def get(self, discount_id: str): + """Get a discount by ID. + https://developer.rechargepayments.com/2021-01/discounts/discounts_retrieve + """ + required_scopes: list[TokenScope] = ["read_discounts"] + self.check_scopes(f"GET /{self.object_list_key}/:discount_id", required_scopes) + + return self.http_get(f"{self.url}/{discount_id}") + + def update(self, discount_id: str, body: DiscountUpdateBody): + """Update a discount. + https://developer.rechargepayments.com/2021-01/discounts/discounts_update + """ + required_scopes: list[TokenScope] = ["write_discounts"] + self.check_scopes(f"PUT /{self.object_list_key}/:discount_id", required_scopes) + + return self.http_put(f"{self.url}/{discount_id}", body) + + def delete(self, discount_id: str): + """Delete a discount. + https://developer.rechargepayments.com/2021-01/discounts/discounts_delete + """ + required_scopes: list[TokenScope] = ["write_discounts"] + self.check_scopes( + f"DELETE /{self.object_list_key}/:discount_id", required_scopes + ) + + return self.http_delete(f"{self.url}/{discount_id}") + + def list(self, query: DiscountListQuery | None = None): + """List discounts. + https://developer.rechargepayments.com/2021-01/discounts/discounts_list + """ + required_scopes: list[TokenScope] = ["read_discounts"] + self.check_scopes(f"GET /{self.object_list_key}", required_scopes) + + return self.http_get(self.url, query) + + def count(self, query: DiscountCountQuery | None = None): + """Receive a count of all discounts. + https://developer.rechargepayments.com/v1#count-discounts + """ + required_scopes: list[TokenScope] = ["read_discounts"] + self.check_scopes(f"GET /{self.object_list_key}/count", required_scopes) + + return self.http_get(f"{self.url}/count", query) diff --git a/recharge/api/metafields.py b/recharge/api/metafields.py new file mode 100644 index 0000000..f48a693 --- /dev/null +++ b/recharge/api/metafields.py @@ -0,0 +1,146 @@ +from recharge.api import RechargeResource +from recharge.api.tokens import TokenScope + +from typing import TypedDict, Required, Literal + +type MetafieldOwnerResource = Literal[ + "address", "store", "customer", "subscription", "order", "charge" +] + +type MetafieldValueType = Literal["string", "json_string", "integer"] + + +class MetafieldCreateBody(TypedDict): + description: str + key: Required[str] + namespace: Required[str] + owner_id: Required[int] + owner_resource: Required[MetafieldOwnerResource] + value: Required[str] + value_type: Required[MetafieldValueType] + + +class MetafieldUpdateBody(TypedDict): + description: str + owner_id: str + owner_resource: MetafieldOwnerResource + value: str + value_type: MetafieldValueType + + +class MetafieldListQuery(TypedDict): + limit: str + namespace: str + owner_id: str + owner_resource: MetafieldOwnerResource + page: str + + +class MetafieldCountQuery(TypedDict): + namespace: str + owner_id: str + owner_resource: MetafieldOwnerResource + + +MetafieldOwnerResourceScopeMap: dict[str, dict[str, TokenScope]] = { + "address": { + "read": "read_customers", + "write": "write_customers", + }, + "store": { + "read": "store_info", + }, + "customer": { + "read": "read_customers", + "write": "write_customers", + }, + "subscription": { + "read": "read_subscriptions", + "write": "write_subscriptions", + }, + "order": { + "read": "read_orders", + "write": "write_orders", + }, + "charge": { + "read": "read_orders", + "write": "write_orders", + }, +} + +type ScopeType = Literal["read", "write"] + + +def resource_scope( + owner_resource: MetafieldOwnerResource, type: ScopeType +) -> TokenScope: + return MetafieldOwnerResourceScopeMap[owner_resource][type] + + +class MetafieldResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/metafields + """ + + object_list_key = "metafields" + + def create(self, body: MetafieldCreateBody): + """Create a metafield. + https://developer.rechargepayments.com/2021-01/metafields/metafields_create + """ + resource = body["owner_resource"] + + required_scopes: list[TokenScope] = [resource_scope(resource, "write")] + self.check_scopes(f"POST /{self.object_list_key}", required_scopes) + + return self.http_post(self.url, body) + + def get(self, metafield_id: str, resource: MetafieldOwnerResource): + """Get a metafield by ID. + https://developer.rechargepayments.com/2021-01/metafields/metafields_retrieve + """ + required_scopes: list[TokenScope] = [resource_scope(resource, "read")] + self.check_scopes(f"GET /{self.object_list_key}/:metafield_id", required_scopes) + + return self.http_get(f"{self.url}/{metafield_id}") + + def update(self, metafield_id: str, body: MetafieldUpdateBody): + """Update a metafield. + https://developer.rechargepayments.com/2021-01/metafields/metafields_update + """ + resource = body["owner_resource"] + required_scopes: list[TokenScope] = [resource_scope(resource, "write")] + self.check_scopes(f"PUT /{self.object_list_key}/:metafield_id", required_scopes) + + return self.http_put(f"{self.url}/{metafield_id}", body) + + def delete(self, metafield_id: str, resource: MetafieldOwnerResource): + """Delete a metafield. + https://developer.rechargepayments.com/2021-01/metafields/metafields_delete + """ + required_scopes: list[TokenScope] = [resource_scope(resource, "write")] + self.check_scopes( + f"DELETE /{self.object_list_key}/:metafield_id", required_scopes + ) + + return self.http_delete(f"{self.url}/{metafield_id}") + + def list(self, query: MetafieldListQuery): + """List metafields. + https://developer.rechargepayments.com/2021-01/metafields/metafields_list + """ + resource = query["owner_resource"] + required_scopes: list[TokenScope] = [resource_scope(resource, "read")] + self.check_scopes(f"GET /{self.object_list_key}", required_scopes) + + return self.http_get(self.url, query) + + def count(self, query: MetafieldCountQuery): + """Retrieve a count of metafields. + https://developer.rechargepayments.com/2021-01/metafields/metafields_count + """ + resource = query["owner_resource"] + required_scopes: list[TokenScope] = [resource_scope(resource, "read")] + self.check_scopes(f"GET /{self.object_list_key}/count", required_scopes) + + return self.http_get(f"{self.url}/count", query) diff --git a/recharge/api/notifications.py b/recharge/api/notifications.py new file mode 100644 index 0000000..a85349c --- /dev/null +++ b/recharge/api/notifications.py @@ -0,0 +1,35 @@ +from typing import Literal, TypedDict + +from recharge.api import RechargeResource +from recharge.api.tokens import TokenScope + +type NotificationTemplateType = Literal["upcoming_charge", "get_account_access"] + + +class NotificationTemplateVars(TypedDict): + address_id: int + charge_id: int + + +class NotificationSendEmailBody(TypedDict): + type: Literal["email"] + template_type: NotificationTemplateType + template_vars: NotificationTemplateVars + + +class NotificationResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/notifications + """ + + def send_email(self, customer_id, body: NotificationSendEmailBody): + """ + Send an email notification to a customer. + https://developer.rechargepayments.com/2021-01/notifications/notifications_get_account_access + """ + required_scopes: list[TokenScope] = ["write_notifications"] + self.check_scopes( + f"POST /customers/{customer_id}/notifications", required_scopes + ) + + return self.http_post(f"{self.url}/customers/{customer_id}/notifications", body) diff --git a/recharge/api/onetimes.py b/recharge/api/onetimes.py new file mode 100644 index 0000000..32e63aa --- /dev/null +++ b/recharge/api/onetimes.py @@ -0,0 +1,96 @@ +from recharge.api.tokens import TokenScope +from . import RechargeResource + +from typing import TypedDict, Required + + +class OnetimeProperty(TypedDict): + name: str + value: str + + +class OnetimeCreateBody(TypedDict): + add_to_next_charge: bool + next_charge_scheduled_at: Required[str] + price: int + product_title: str + properties: list[OnetimeProperty] + quantity: Required[int] + shopify_product_id: int + shopify_variant_id: Required[int] + + +class OnetimeUpdateBody(TypedDict): + next_charge_scheduled_at: str + price: int + product_title: str + properties: list[OnetimeProperty] + quantity: int + shopify_product_id: int + sku: str + shopify_variant_id: int + + +class OnetimeListQuery(TypedDict): + address_id: str + created_at_max: str + created_at_min: str + customer_id: str + limit: str + page: str + shopify_customer_id: str + updated_at_max: str + updated_at_min: str + + +class OnetimeResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/onetimes + """ + + object_list_key = "onetimes" + + def create(self, body: OnetimeCreateBody): + """Create a Onetime + https://developer.rechargepayments.com/2021-01/onetimes/onetimes_create + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes("POST /onetimes", required_scopes) + + return self.http_post(self.url, body) + + def get(self, onetime_id: str): + """Get a Onetime + https://developer.rechargepayments.com/2021-01/onetimes/onetimes_retrieve + """ + required_scopes: list[TokenScope] = ["read_subscriptions"] + self.check_scopes("GET /onetimes/:onetime_id", required_scopes) + + return self.http_get(f"{self.url}/{onetime_id}") + + def update(self, onetime_id: str, body: OnetimeUpdateBody): + """Update a Onetime + https://developer.rechargepayments.com/2021-01/onetimes/onetimes_update + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes("PUT /onetimes/:onetime_id", required_scopes) + + return self.http_put(f"{self.url}/{onetime_id}", body) + + def delete(self, onetime_id: str): + """Delete a Onetime. + https://developer.rechargepayments.com/2021-01/onetimes/onetimes_delete + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes("DELETE /onetimes/:onetime_id", required_scopes) + + return self.http_delete(f"{self.url}/{onetime_id}") + + def list(self, query: OnetimeListQuery): + """List Onetimes. + https://developer.rechargepayments.com/2021-01/onetimes/onetimes_list + """ + required_scopes: list[TokenScope] = ["read_subscriptions"] + self.check_scopes("GET /onetimes", required_scopes) + + return self.http_get(self.url, query) diff --git a/recharge/api/orders.py b/recharge/api/orders.py new file mode 100644 index 0000000..81eb23a --- /dev/null +++ b/recharge/api/orders.py @@ -0,0 +1,194 @@ +from recharge.api import RechargeResource +from recharge.api.tokens import TokenScope + +from typing import TypedDict, Literal + + +class OrderBillingAddress(TypedDict): + address1: str + province: str + address2: str + city: str + company: str + country: str + first_name: str + last_name: str + phone: str + zip: str + + +class OrderShippingAddress(TypedDict): + address1: str + province: str + address2: str + city: str + company: str + country: str + first_name: str + last_name: str + phone: str + zip: str + + +class OrderCustomer(TypedDict): + first_name: str + last_name: str + email: str + + +class OrderUpdateBody(TypedDict): + billing_address: OrderBillingAddress + shipping_address: OrderShippingAddress + customer: OrderCustomer + + +type OrderStatus = Literal["SUCCESS", "QUEUED", "ERROR", "REFUNDED", "SKIPPED"] + + +class OrderListQuery(TypedDict): + address_id: str + charge_id: str + created_at_max: str + created_at_min: str + customer_id: str + ids: str + limit: str + page: str + scheduled_at_max: str + scheduled_at_min: str + shipping_date: str + shopify_order_id: str + has_external_id: str + status: OrderStatus + subscription_id: str + updated_at_max: str + updated_at_min: str + + +class OrderCountQuery(TypedDict): + address_id: str + charge_id: str + created_at_max: str + created_at_min: str + customer_id: str + scheduled_at_max: str + scheduled_at_min: str + shopify_customer_id: str + status: OrderStatus + subscription_id: str + updated_at_max: str + updated_at_min: str + + +class OrderChangeDateBody(TypedDict): + scheduled_at: str + + +class OrderChangeVariantBody(TypedDict): + new_shopify_variant_id: str + + +class OrderCloneBody(TypedDict): + scheduled_at: str + + +class OrderResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/orders + """ + + object_list_key = "orders" + + def get(self, order_id: str): + """Get an order. + https://developer.rechargepayments.com/2021-01/orders/orders_retrieve + """ + required_scopes: list[TokenScope] = ["read_orders"] + self.check_scopes(f"GET /orders/{order_id}", required_scopes) + + return self.http_get(f"{self.url}/{order_id}") + + def update(self, order_id: str, body: OrderUpdateBody): + """Update an order. + https://developer.rechargepayments.com/2021-01/orders/orders_update + """ + required_scopes: list[TokenScope] = ["write_orders"] + self.check_scopes(f"PUT /orders/{order_id}", required_scopes) + + return self.http_put(f"{self.url}/{order_id}", body) + + def delete(self, order_id: str): + """Delete an order. + https://developer.rechargepayments.com/2021-01/orders/orders_delete + """ + required_scopes: list[TokenScope] = ["write_orders"] + self.check_scopes(f"DELETE /orders/{order_id}", required_scopes) + + return self.http_delete(f"{self.url}/{order_id}") + + def list(self, query: OrderListQuery): + """List orders. + https://developer.rechargepayments.com/2021-01/orders/orders_list + """ + required_scopes: list[TokenScope] = ["read_orders"] + self.check_scopes("GET /orders", required_scopes) + + return self.http_get(self.url, query) + + def count(self, query: OrderCountQuery): + """Count orders. + https://developer.rechargepayments.com/2021-01/orders/orders_count + """ + required_scopes: list[TokenScope] = ["read_orders"] + self.check_scopes("GET /orders/count", required_scopes) + + return self.http_get(f"{self.url}/count", query) + + def change_date(self, order_id: str, body: OrderChangeDateBody): + """Change the date of a queued order. + https://developer.rechargepayments.com/2021-01/orders/orders_change_date + """ + required_scopes: list[TokenScope] = ["write_orders"] + self.check_scopes("POST /orders/:order_id/change_date", required_scopes) + + return self.http_post( + f"{self.url}/{order_id}/change_date", + body, + ) + + def change_variant( + self, order_id: str, old_variant_id: str, body: OrderChangeVariantBody + ): + """Change an order variant. + https://developer.rechargepayments.com/v1#change-an-order-variant + """ + required_scopes: list[TokenScope] = ["write_orders"] + self.check_scopes( + "PUT /orders/:order_id/update_shopify_variant/:old_variant_id", + required_scopes, + ) + + return self.http_put( + f"{self.url}/{order_id}/update_shopify_variant/{old_variant_id}", body + ) + + def clone(self, order_id: str, charge_id: str, body: OrderCloneBody): + """Clone an order. + https://developer.rechargepayments.com/2021-01/orders/orders_clone + """ + required_scopes: list[TokenScope] = ["write_orders"] + self.check_scopes("POST /orders/:order_id/clone", required_scopes) + + return self.http_post( + f"{self.url}/clone_order_on_success_charge/{order_id}/charge/{charge_id}", + body, + ) + + def delay(self, order_id: str): + """Delay an order. + https://developer.rechargepayments.com/2021-01/orders/orders_delay + """ + required_scopes: list[TokenScope] = ["write_orders"] + self.check_scopes("POST /orders/:order_id/delay", required_scopes) + + return self.http_post(f"{self.url}/{order_id}/delay", None) diff --git a/recharge/api/products.py b/recharge/api/products.py new file mode 100644 index 0000000..d5ea612 --- /dev/null +++ b/recharge/api/products.py @@ -0,0 +1,141 @@ +from recharge.api.tokens import TokenScope +from recharge.api import RechargeResource + +from typing import TypedDict, Required, Literal + +type ProductDiscountType = Literal["percentage"] + +type ProductStorefrontPurchaseOptions = Literal[ + "subscription_only", "subscription_and_onetime" +] + + +class ProductCreateBody(TypedDict): + charge_interval_frequency: int + cutoff_day_of_month: int + cutoff_day_of_week: int + discount_amount: str + discount_type: ProductDiscountType + expire_after_specific_number_of_charges: str + modifiable_properties: list[str] + order_day_of_month: int + order_day_of_week: int + order_interval_frequency_options: list[int] + shopify_product_id: Required[int] + storefront_purchase_options: ProductStorefrontPurchaseOptions + + +class ProductImages(TypedDict): + large: str + medium: str + original: str + small: str + + +type ProductOrderIntervalUnit = Literal["day", "week", "month"] + + +class ProductGetQuery(TypedDict): + charge_interval_frequency: str + created_at: str + cutoff_day_of_month: str + cutoff_day_of_week: str + discount_amount: str + discount_type: ProductDiscountType + expire_after_specific_number_of_charges: str + handle: str + images: ProductImages + modifiable_properties: list[str] + number_charges_until_expiration: str + order_day_of_month: str + order_day_of_week: str + order_interval_frequency: str + order_interval_unit: ProductOrderIntervalUnit + shopify_product_id: str + storefront_purchase_options: ProductStorefrontPurchaseOptions + title: str + updated_at: str + + +class ProductUpdateBody(TypedDict): + charge_interval_frequency: int + cutoff_day_of_month: int + cutoff_day_of_week: int + discount_amount: str + discount_type: ProductDiscountType + expire_after_specific_number_of_charges: str + modifiable_properties: list[str] + order_day_of_month: int + order_day_of_week: int + order_interval_unit: ProductOrderIntervalUnit + shopify_product_id: Required[int] + storefront_purchase_options: ProductStorefrontPurchaseOptions + + +class ProductListQuery(TypedDict): + id: str + limit: str + shopify_product_id: int + page: str + + +class ProductResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/products + """ + + object_list_key = "products" + + def create(self, body: ProductCreateBody): + """Create a product. + https://developer.rechargepayments.com/2021-01/products/products_create + """ + required_scopes: list[TokenScope] = ["write_products"] + self.check_scopes("POST /products", required_scopes) + + return self.http_post(self.url, body) + + def get(self, product_id: str): + """Get a product. + https://developer.rechargepayments.com/2021-01/products/products_retrieve + """ + required_scopes: list[TokenScope] = ["read_products"] + self.check_scopes(f"GET /products/{product_id}", required_scopes) + + return self.http_get(f"{self.url}/{product_id}") + + def update(self, product_id: str, body: ProductUpdateBody): + """Update a product. + https://developer.rechargepayments.com/2021-01/products/products_update + """ + required_scopes: list[TokenScope] = ["write_products"] + self.check_scopes(f"PUT /products/{product_id}", required_scopes) + + return self.http_put(f"{self.url}/{product_id}", body) + + def delete(self, product_id: str): + """Delete a product. + https://developer.rechargepayments.com/2021-01/products/products_delete + """ + required_scopes: list[TokenScope] = ["write_products"] + self.check_scopes(f"DELETE /products/{product_id}", required_scopes) + + return self.http_delete(f"{self.url}/{product_id}") + + def list(self, query: ProductListQuery): + """List products. + https://developer.rechargepayments.com/2021-01/products/products_list + """ + required_scopes: list[TokenScope] = ["read_products"] + self.check_scopes("GET /products", required_scopes) + + return self.http_get(self.url, query) + + def count(self): + """Count products. + https://developer.rechargepayments.com/2021-01/products/products_count + """ + required_scopes: list[TokenScope] = ["read_products"] + self.check_scopes("GET /products/count", required_scopes) + + return self.http_get(f"{self.url}/count") diff --git a/recharge/api/shop.py b/recharge/api/shop.py new file mode 100644 index 0000000..c75b8d9 --- /dev/null +++ b/recharge/api/shop.py @@ -0,0 +1,28 @@ +from recharge.api.tokens import TokenScope +from recharge.api import RechargeResource + + +class ShopResource(RechargeResource): + """ + https://developer.rechargepayments.com/v1#shop + """ + + object_list_key = "shop" + + def get(self): + """Retrieve store using the Recharge API. + https://developer.rechargepayments.com/2021-01/shop/shop_retrieve + """ + required_scopes: list[TokenScope] = ["store_info"] + self.check_scopes("GET /shop", required_scopes) + + return self.http_get(f"{self.url}") + + def shipping_countries(self): + """Retrieve shipping countries where items can be shipped. + https://developer.rechargepayments.com/2021-01/shop/shop_shipping_countries + """ + required_scopes: list[TokenScope] = ["store_info"] + self.check_scopes("GET /shipping_countries", required_scopes) + + return self.http_get(f"{self.url}/shipping_countries") diff --git a/recharge/api/subscriptions.py b/recharge/api/subscriptions.py new file mode 100644 index 0000000..7b47a17 --- /dev/null +++ b/recharge/api/subscriptions.py @@ -0,0 +1,250 @@ +from recharge.api import RechargeResource +from recharge.api.tokens import TokenScope + +from typing import TypedDict, Literal, Required + +type SubscriptionOrderIntervalUnit = Literal["day", "week", "month"] + + +class SubscriptionProperty(TypedDict): + name: str + value: str + + +type SubscriptionStatus = Literal["ACTIVE", "CANCELLED", "EXPIRED"] + + +class SubscriptionCreateBody(TypedDict): + address_id: Required[int] + charge_interval_frequency: Required[int] + customer_id: str + expire_after_specific_number_of_charges: str + next_charge_scheduled_at: Required[str] + order_day_of_month: str + order_day_of_week: str + order_interval_frequency: Required[str] + order_interval_unit: Required[SubscriptionOrderIntervalUnit] + price: str + properties: list[SubscriptionProperty] + product_title: str + quantity: Required[str] + shopify_product_id: Required[str] + shopify_variant_id: Required[str] + status: SubscriptionStatus + + +class SubscriptionUpdateBody(TypedDict): + charge_interval_frequency: str + commit_update: bool + expire_after_specific_number_of_charges: str + order_day_of_month: str + order_day_of_week: str + order_interval_frequency: str + order_interval_unit: SubscriptionOrderIntervalUnit + override: str + price: str + product_title: str + properties: list[SubscriptionProperty] + quantity: str + shopify_variant_id: str + sku: str + sku_override: str + use_shopify_variant_defaults: bool + variant_title: str + + +class SubscriptionDeleteBody(TypedDict): + send_email: bool + + +class SubscriptionListQuery(TypedDict): + address_id: str + created_at_max: str + created_at_min: str + customer_id: str + ids: str + include_onetimes: str + limit: str + page: str + shopify_customer_id: str + shopify_variant_id: str + status: SubscriptionStatus + updated_at_max: str + updated_at_min: str + + +class SubscriptionCountQuery(TypedDict): + address_id: str + created_at_max: str + created_at_min: str + customer_id: str + shopify_customer_id: str + shopify_variant_id: str + status: SubscriptionStatus + updated_at_max: str + updated_at_min: str + + +class SubscriptionChangeDateBody(TypedDict, total=True): + date: str + + +class SubscriptionChangeAddressBody(TypedDict): + address_id: Required[str] + next_charge_scheduled_at: str + + +class SubscriptionCancelBody(TypedDict): + cancellation_reason: Required[str] + cancellation_reason_comments: str + send_email: bool + + +class SubscriptionBulkCreateBody(TypedDict): + subscriptions: list[SubscriptionCreateBody] + + +class SubscriptionBulkUpdateBody(TypedDict): + subscriptions: list[SubscriptionUpdateBody] + + +class SubscriptionBulkDeleteBodyInner(TypedDict): + id: Required[str] + send_email: bool + + +class SubscriptionBulkDeleteBody(TypedDict): + subscriptions: list[SubscriptionBulkDeleteBodyInner] + + +class SubscriptionResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/subscriptions + """ + + object_list_key = "subscriptions" + + def create(self, body: SubscriptionCreateBody): + """Create a subscription. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_create + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes("POST /subscriptions", required_scopes) + + return self.http_post(self.url, body) + + def get(self, subscription_id: str): + """Get a subscription. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_retrieve + """ + required_scopes: list[TokenScope] = ["read_subscriptions"] + self.check_scopes("GET /subscriptions/:subscription_id", required_scopes) + + return self.http_get(f"{self.url}/{subscription_id}") + + def update(self, subscription_id: str, body: SubscriptionUpdateBody): + """Update a subscription. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_update + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes("PUT /subscriptions/:subscription_id", required_scopes) + + return self.http_put(f"{self.url}/{subscription_id}", body) + + def delete(self, subscription_id: str, body: SubscriptionDeleteBody): + """Delete a subscription. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_delete + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes("DELETE /subscriptions/:subscription_id", required_scopes) + + return self.http_delete(f"{self.url}/{subscription_id}", body) + + def list(self, query: SubscriptionListQuery): + """List subscriptions. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_list + """ + required_scopes: list[TokenScope] = ["read_subscriptions"] + self.check_scopes("GET /subscriptions", required_scopes) + + return self.http_get(self.url, query) + + def count(self, query: SubscriptionCountQuery): + """Count subscriptions. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_count + """ + required_scopes: list[TokenScope] = ["read_subscriptions"] + self.check_scopes("GET /subscriptions/count", required_scopes) + + return self.http_get(f"{self.url}/count", query) + + def change_date(self, subscription_id: str, body: SubscriptionChangeDateBody): + """Change the date of a queued subscription. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_change_date + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes( + "POST /subscriptions/:subscription_id/change_date", required_scopes + ) + + return self.http_post(f"{self.url}/{subscription_id}/change_date", body) + + def change_address(self, subscription_id: str, body: SubscriptionChangeAddressBody): + """Change the address of a subscription. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_change_address + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes( + "POST /subscriptions/:subscription_id/change_address", required_scopes + ) + + return self.http_post(f"{self.url}/{subscription_id}/change_address", body) + + def cancel(self, subscription_id: str): + """Cancel a subscription. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_cancel + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes( + "POST /subscriptions/:subscription_id/cancel", required_scopes + ) + + return self.http_post(f"{self.url}/{subscription_id}/cancel", {}) + + def activate(self, subscription_id: str): + """Activate a cancelled subscription. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_activate + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes( + "POST /subscriptions/:subscription_id/activate", required_scopes + ) + + return self.http_post(f"{self.url}/{subscription_id}/activate", {}) + + def bulk_create(self, body: SubscriptionBulkCreateBody): + """Bulk create subscriptions. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_bulk_create + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes("POST /subscriptions/bulk_create", required_scopes) + + return self.http_post(f"{self.url}/bulk_create", body) + + def bulk_update(self, body: SubscriptionBulkUpdateBody): + """Bulk update subscriptions. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_bulk_update + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes("POST /subscriptions/bulk_update", required_scopes) + + return self.http_post(f"{self.url}/bulk_update", body) + + def bulk_delete(self, body: SubscriptionBulkDeleteBody): + """Bulk delete subscriptions. + https://developer.rechargepayments.com/2021-01/subscriptions/subscriptions_bulk_delete + """ + required_scopes: list[TokenScope] = ["write_subscriptions"] + self.check_scopes("POST /subscriptions/bulk_delete", required_scopes) + + return self.http_post(f"{self.url}/bulk_delete", body) diff --git a/recharge/api/tokens.py b/recharge/api/tokens.py new file mode 100644 index 0000000..489974c --- /dev/null +++ b/recharge/api/tokens.py @@ -0,0 +1,61 @@ +from recharge.api import RechargeResource + +from typing import TypedDict, Literal + + +class TokenClient(TypedDict): + name: str + email: str + + +type TokenScope = Literal[ + "write_orders", + "read_orders", + "write_discounts", + "read_discounts", + "write_subscriptions", + "read_subscriptions", + "write_payments", + "read_payments", + "write_payment_methods", + "read_payment_methods", + "write_customers", + "read_customers", + "write_products", + "read_products", + "store_info", + "write_batches", + "read_batches", + "read_accounts", + "write_checkouts", + "read_checkouts", + "write_notifications", + "read_events", + "write_retention_strategies", + "read_gift_purchases", + "write_gift_purchases", + "read_gift_purchases", + "write_gift_purchases", + "read_bundle_products", +] + + +class TokenInformation(TypedDict): + client: TokenClient + contact_email: str + name: str + scopes: list[TokenScope] + + +class TokenResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/token_information/token_information_object + """ + + object_list_key = "tokens_information" + + def get(self) -> TokenInformation: + """Get token information. + https://developer.rechargepayments.com/2021-01/token_information/token_information_retrieve + """ + return self.http_get(f"{self.url}") diff --git a/recharge/api/webhooks.py b/recharge/api/webhooks.py new file mode 100644 index 0000000..2467672 --- /dev/null +++ b/recharge/api/webhooks.py @@ -0,0 +1,123 @@ +from recharge.api.tokens import TokenScope +from . import RechargeResource + +from typing import Literal, Required, TypedDict + +type WebhookTopic = Literal[ + "address/created", + "address/updated", + "async_batch/processed", + "bundle_selection/created", + "bundle_selection/updated", + "bundle_selection/deleted", + "customer/activated", + "customer/created", + "customer/deactivated", + "customer/payment_method_updated", + "customer/updated", + "customer/deleted", + "charge/created", + "charge/failed", + "charge/max_retries_reached", + "charge/paid", + "charge/refunded", + "charge/uncaptured", + "charge/upcoming", + "charge/updated", + "charge/deleted", + "checkout/created", + "checkout/completed", + "checkout/processed", + "checkout/updated", + "onetime/created", + "onetime/deleted", + "onetime/updated", + "order/cancelled", + "order/created", + "order/deleted", + "order/processed", + "order/payment_captured", + "order/upcoming", + "order/updated", + "order/success", + "plan/created", + "plan/deleted", + "plan/updated", + "subscription/activated", + "subscription/cancelled", + "subscription/created", + "subscription/deleted", + "subscription/skipped", + "subscription/updated", + "subscription/unskipped", + "subscription/swapped", + "subscription/paused", + "store/updated", + "recharge/uninstalled", +] +WebhookTopicMap: dict[str, TokenScope] = { + "address": "read_customers", + "async_batch": "read_batches", + "bundle_selection": "read_subscriptions", + "customer": "read_customers", + "charge": "read_orders", + "checkout": "read_checkouts", + "onetime": "read_subscriptions", + "order": "read_orders", + "product": "read_products", + "subscription": "read_subscriptions", + "shop": "store_info", + "recharge": "store_info", +} + +type WebhookIncludedObject = Literal[ + "addresses", "collections", "customer", "metafields" +] + + +class WebhookCreateBody(TypedDict): + address: Required[str] + inclded_objects: list[WebhookIncludedObject] + topic: Required[WebhookTopic] + + +class WebhookResource(RechargeResource): + """ + https://developer.rechargepayments.com/2021-01/webhooks_endpoints/webhooks_object + """ + + object_list_key = "webhooks" + + def create(self, body: WebhookCreateBody): + """Create a webhook. + https://developer.rechargepayments.com/2021-01/webhooks_endpoints/webhooks_create + """ + resource = body["topic"].split("/")[0] + required_scopes: list[TokenScope] = [WebhookTopicMap[resource]] + self.check_scopes("POST /webhooks", required_scopes) + + return self.http_post(self.url, body) + + def get(self, webhook_id: str): + """Get a webhook. + https://developer.rechargepayments.com/2021-01/webhooks_endpoints/webhooks_retrieve + """ + return self.http_get(f"{self.url}/{webhook_id}") + + def update(self, webhook_id: str): + """Update a webhook. + https://developer.rechargepayments.com/2021-01/webhooks_endpoints/webhooks_update + """ + return self.http_delete(f"{self.url}/{webhook_id}") + + def list(self): + """List webhooks. + https://developer.rechargepayments.com/2021-01/webhooks_endpoints/webhooks_list + """ + return self.http_get(self.url) + + def test(self, webhook_id: str): + """Test a webhook. + https://developer.rechargepayments.com/2021-01/webhooks_endpoints/webhooks_test + """ + return self.http_post(f"{self.url}/{webhook_id}/test") diff --git a/recharge/resources.py b/recharge/resources.py deleted file mode 100644 index 9ad4478..0000000 --- a/recharge/resources.py +++ /dev/null @@ -1,464 +0,0 @@ -import logging -import time - -from urllib.parse import urlencode - -import requests - -log = logging.getLogger(__name__) - - -class RechargeResource(object): - """ - Resource from the Recharge API. This class handles - logging, sending requests, parsing JSON, and rate - limiting. - - Refer to the API docs to see the expected responses. - https://developer.rechargepayments.com/ - """ - base_url = 'https://api.rechargeapps.com' - object_list_key = None - - def __init__(self, access_token=None, log_debug=False): - self.log_debug = log_debug - self.headers = { - 'Accept': 'application/json', - 'Content-Type': 'application/json', - 'X-Recharge-Access-Token': access_token, - } - - def log(self, url, response): - if self.log_debug: - log.info(url) - log.info(response.headers['X-Recharge-Limit']) - - @property - def url(self): - return f'{self.base_url}/{self.object_list_key}' - - def http_delete(self, url): - response = requests.delete(url, headers=self.headers) - log.info(url) - log.info(response.headers['X-Recharge-Limit']) - if response.status_code == 429: - return self.http_delete(url) - return response - - def http_get(self, url): - response = requests.get(url, headers=self.headers) - self.log(url, response) - if response.status_code == 429: - time.sleep(1) - return self.http_get(url) - return response.json() - - def http_put(self, url, data): - response = requests.put(url, json=data, headers=self.headers) - self.log(url, response) - if response.status_code == 429: - time.sleep(1) - return self.http_put(url, data) - return response.json() - - def http_post(self, url, data): - response = requests.post(url, json=data, headers=self.headers) - self.log(url, response) - if response.status_code == 429: - time.sleep(1) - return self.http_post(url, data) - return response.json() - - def create(self, data): - return self.http_post(self.url, data) - - def update(self, resource_id, data): - return self.http_put(f'{self.url}/{resource_id}', data) - - def get(self, resource_id): - return self.http_get(f'{self.url}/{resource_id}') - - def list(self, url_params=None): - """ - The list method takes a dictionary of filter parameters. - Refer to the recharge docs for available filters for - each resource. - """ - params = ('?' + urlencode(url_params, doseq=True)) if url_params else '' - return self.http_get(self.url + params) - - -class RechargeAddress(RechargeResource): - """ - https://developer.rechargepayments.com/#addresses - """ - object_list_key = 'addresses' - - def apply_discount(self, address_id, discount_code): - """ Apply a discount code to an address. - https://developer.rechargepayments.com/#add-discount-to-address-new - """ - return self.http_post( - f'{self.url}/{address_id}/apply_discount', - {'discount_code': discount_code} - ) - - def count(self, data=None): - """Retrieve the count of addresses. - https://developer.rechargepayments.com/v1#count-addresses - """ - params = ('?' + urlencode(data, doseq=True)) if data else '' - return self.http_get(f'{self.url}/count{params}') - - def create(self, customer_id, data): - """Create an address for the customer. - https://developer.rechargepayments.com/#create-address - """ - url = f'{self.base_url}/customers/{customer_id}/{self.object_list_key}' - return self.http_post(url, data) - - def delete(self, address_id): - """Delete an address. - https://developer.rechargepayments.com/v1#delete-an-address - """ - return self.http_delete(f'{self.url}/{address_id}') - - -class RechargeCharge(RechargeResource): - """ - https://developer.rechargepayments.com/#charges - """ - object_list_key = 'charges' - - def change_next_charge_date(self, charge_id, to_date): - """Change the date of a queued charge. - https://developer.rechargepayments.com/#change-next-charge-date - """ - return self.http_put( - f'{self.url}/{charge_id}/change_next_charge_date', - {'next_charge_date': to_date} - ) - - def count(self, data=None): - """Retrieve a count of charges. - https://developer.rechargepayments.com/v1#count-charges - """ - params = ('?' + urlencode(data, doseq=True)) if data else '' - return self.http_get(f'{self.url}/count{params}') - - def full_refund(self, charge_id): - """Full refund a charge - https://developer.rechargepayments.com/v1#full-refund-a-charge - """ - return self.http_post( - f'{self.url}/{charge_id}/refund', - {"full_refund": True} - ) - - def process(self, charge_id): - """Process a charge. - https://developer.rechargepayments.com/v1#process-a-charge - """ - return self.http_post( - f'{self.url}/{charge_id}/process', - {} - ) - - def refund(self, charge_id, amount): - """Refund a charge. - https://developer.rechargepayments.com/v1#refund-a-charge - """ - return self.http_post( - f'{self.url}/{charge_id}/refund', - {"amount": amount} - ) - - def skip(self, charge_id, subscription_id): - """Skip a charge. - https://developer.rechargepayments.com/v1#skip-a-charge - """ - return self.http_post( - f'{self.url}/{charge_id}/skip', - {"subscription_id": subscription_id} - ) - - def unskip(self, charge_id, subscription_id): - """Unskip a charge. - https://developer.rechargepayments.com/v1#unskip-a-charge - """ - return self.http_post( - f'{self.url}/{charge_id}/unskip', - {"subscription_id": subscription_id} - ) - - -class RechargeCheckout(RechargeResource): - """ - https://developer.rechargepayments.com/#checkouts - """ - object_list_key = 'checkouts' - - def charge(self, checkout_id, data): - """Process (charge) a checkout. - https://developer.rechargepayments.com/#process-checkout-beta - """ - return self.http_post( - f'{self.url}/{checkout_id}/charge', - data - ) - - def get_shipping(self, checkout_id): - """Retrieve shipping rates for a checkout - https://developer.rechargepayments.com/v1#retrieve-shipping-rates-for-a-checkout - """ - return self.http_get(f'{self.url}/{checkout_id}/shipping_rates') - - -class RechargeCustomer(RechargeResource): - """ - https://developer.rechargepayments.com/#customers - """ - object_list_key = 'customers' - - def count(self, data=None): - """Retrieve a count of customers. - https://developer.rechargepayments.com/v1#count-customers - """ - params = ('?' + urlencode(data, doseq=True)) if data else '' - return self.http_get(f'{self.url}/count{params}') - - def delete(self, customer_id): - """Delete a customer. - https://developer.rechargepayments.com/v1#delete-a-customer - """ - return self.http_delete(f'{self.url}/{customer_id}') - - -class RechargeOrder(RechargeResource): - """ - https://developer.rechargepayments.com/#orders - """ - object_list_key = 'orders' - - def change_date(self, order_id, to_date): - """Change the date of a queued order. - https://developer.rechargepayments.com/#change-order-date - """ - return self.http_put( - f'{self.url}/{order_id}/change_date', - {'scheduled_at': f'{to_date}T00:00:00'} - ) - - def change_variant(self, order_id, old_variant_id, new_variant_id): - """Change an order variant. - https://developer.rechargepayments.com/v1#change-an-order-variant - """ - return self.http_put( - f'{self.url}/{order_id}/update_shopify_variant/{old_variant_id}', - { - "new_shopify_variant_id": new_variant_id, - "shopify_variant_id": old_variant_id - } - ) - - def count(self, data=None): - """Retrieve a count of all orders. - https://developer.rechargepayments.com/v1#count-orders - """ - params = ('?' + urlencode(data, doseq=True)) if data else '' - return self.http_get(f'{self.url}/count{params}') - - def delete(self, order_id): - """https://developer.rechargepayments.com/v1#delete-an-order - https://developer.rechargepayments.com/#delete-order-beta - """ - return self.http_delete(f'{self.url}/{order_id}') - - def update_items(self, order_id, data): - """Update order line_items - https://developer.rechargepayments.com/v1#update-order-line_items - """ - return self.http_put( - f'{self.url}/{order_id}', - data - ) - - -class RechargeSubscription(RechargeResource): - """ - https://developer.rechargepayments.com/#subscriptions - """ - object_list_key = 'subscriptions' - - def activate(self, subscription_id): - """Activate a cancelled subscription. - https://developer.rechargepayments.com/v1#activate-a-subscription - """ - return self.http_post( - f'{self.url}/{subscription_id}/activate', - {} - ) - - def cancel(self, subscription_id, data=None): - """Cancel a subscription. - https://developer.rechargepayments.com/#cancel-subscription - """ - return self.http_post(f'{self.url}/{subscription_id}/cancel', data) - - def change_address(self, subscription_id, address_id): - """Change a subscription address. - https://developer.rechargepayments.com/v1#change-a-subscription-address - """ - return self.http_post( - f'{self.url}/{subscription_id}/change_address', - {"address_id": address_id} - ) - - def count(self, data=None): - """Returns the count of subscriptions. - https://developer.rechargepayments.com/v1#count-subscriptions - """ - params = ('?' + urlencode(data, doseq=True)) if data else '' - return self.http_get(f'{self.url}/count{params}') - - def delete(self, subscription_id, data=None): - """Delete a subscription - https://developer.rechargepayments.com/#delete-subscription - """ - return self.http_delete(f'{self.url}/{subscription_id}') - - def set_next_charge_date(self, subscription_id, date): - """Change the next charge date of a subscription - https://developer.rechargepayments.com/#change-next-charge-date-on-subscription - """ - return self.http_post( - f'{self.url}/{subscription_id}/set_next_charge_date', - {'date': date} - ) - - -class RechargeOnetime(RechargeResource): - """ - https://developer.rechargepayments.com/#onetimes - """ - object_list_key = 'onetimes' - - def delete(self, onetime_id, data=None): - """Delete a Onetime - https://developer.rechargepayments.com/#delete-a-onetime - """ - return self.http_delete(f'{self.url}/{onetime_id}') - - -class RechargeDiscount(RechargeResource): - """ - https://developer.rechargepayments.com/#discounts - """ - object_list_key = 'discounts' - - def apply(self, resource, resource_id, discount_code): - """Apply a discount to an address or charge. - https://developer.rechargepayments.com/v1#apply-a-discount-to-an-address - """ - if resource not in ['addresses', 'charges']: - raise ValueError("Resource is not 'addresses' or 'charges'") - return self.http_post( - f'{self.base_url}/{resource}/{resource_id}/apply_discount', - {"discount_code": discount_code} - ) - - def count(self, data=None): - """Receive a count of all discounts. - https://developer.rechargepayments.com/v1#count-discounts - """ - params = ('?' + urlencode(data, doseq=True)) if data else '' - return self.http_get(f'{self.url}/count{params}') - - def delete(self, discount_id, data=None): - """Delete a Discount - https://developer.rechargepayments.com/#delete-a-discount - """ - return self.http_delete(f'{self.url}/{discount_id}') - - def remove(self, resource, resource_id): - """Remove a discount from a charge or address without destroying the discount. - https://developer.rechargepayments.com/v1#remove-a-discount - """ - if resource not in ['addresses', 'charges']: - raise ValueError("Resource is not 'addresses' or 'charges'.") - return self.http_post( - f'{self.base_url}/{resource}/{resource_id}/remove_discount', - {} - ) - - -class RechargeProduct(RechargeResource): - """ - https://developer.rechargepayments.com/v1#products - """ - object_list_key = 'products' - - def count(self, data=None): - """Retrieve a count of all products. - https://developer.rechargepayments.com/v1#count-products - """ - params = ('?' + urlencode(data, doseq=True)) if data else '' - return self.http_get(f'{self.url}/count{params}') - - def delete(self, product_id): - """Delete a product from store. - https://developer.rechargepayments.com/v1#delete-a-product - """ - return self.http_delete(f'{self.url}/{product_id}') - - -class RechargeShop(RechargeResource): - """ - https://developer.rechargepayments.com/v1#shop - """ - object_list_key = 'shop' - - def retrieve(self): - """Retrieve store using the Recharge API. - https://developer.rechargepayments.com/v1#retrieve-a-shop - """ - return self.http_get(f'{self.url}') - - def shipping_countries(self): - """Retrieve shipping countries where items can be shipped. - https://developer.rechargepayments.com/v1#retrieve-a-shop - """ - return self.http_get(f'{self.url}/shipping_countries') - - -class RechargeMetafield(RechargeResource): - """ - https://developer.rechargepayments.com/v1#the-metafield-object - """ - object_list_key = 'metafields' - - def count(self, data=None): - """Retrieves a number of metafields for some specific query parameter. - https://developer.rechargepayments.com/v1#count-metafields - """ - params = ('?' + urlencode(data, doseq=True)) if data else '' - return self.http_get(f'{self.url}/count{params}') - - def delete(self, metafield_id): - """Delete a metafield. - https://developer.rechargepayments.com/v1#delete-a-metafield - """ - return self.http_delete(f'{self.url}/{metafield_id}') - - -class RechargeWebhook(RechargeResource): - """ - https://developer.rechargepayments.com/v1#webhooks - """ - object_list_key = 'webhooks' - - def delete(self, webhook_id): - """Delete a webhook. - https://developer.rechargepayments.com/v1#delete-a-webhook - """ - return self.http_delete(f'{self.url}/{webhook_id}') diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..8e637bf --- /dev/null +++ b/requirements.txt @@ -0,0 +1,11 @@ +# This file was autogenerated by uv via the following command: +# uv pip compile - -o requirements.txt +certifi==2024.2.2 + # via requests +charset-normalizer==3.3.2 + # via requests +idna==3.7 + # via requests +requests==2.31.0 +urllib3==2.2.1 + # via requests diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..15ea301 --- /dev/null +++ b/setup.cfg @@ -0,0 +1,28 @@ +[metadata] +name = recharge-api +version = 1.4.1 +author = ChemicalLuck +description = Python API Wrapper for Recharge +long_description = file: README.md +url = http://github.com/ChemicalLuck/recharge-api +license = MIT +keywords = api recharge +classifiers = + Development Status :: 3 - Alpha + Intended Audience :: Information Technology + License :: OSI Approved :: MIT License + Natural Language :: English + Programming Language :: Python :: 3 + Topic :: Software Development :: Libraries :: Python Modules + Topic :: Software Development :: Testing + Topic :: Utilities + + +[options] +zip_safe = False +include_package_data = True +packages = find: +install_requires = + requests + +python_requires = >=3.6 diff --git a/setup.py b/setup.py deleted file mode 100644 index 3aae061..0000000 --- a/setup.py +++ /dev/null @@ -1,39 +0,0 @@ - -import codecs -import os -import re -from setuptools import setup - - -here = os.path.abspath(os.path.dirname(__file__)) - -with open("README.rst", "r", encoding="utf-8") as fh: - long_description = fh.read() - -setup( - name='recharge-api', - version='1.4.1', - author='ChemicalLuck', - author_email='chemicalluck@outlook.com', - packages=['recharge'], - include_package_data=True, - url='http://github.com/ChemicalLuck/recharge-api', - license='MIT', - description='Python API Wrapper for Recharge', - classifiers=[ - 'Development Status :: 3 - Alpha', - 'Intended Audience :: Information Technology', - 'License :: OSI Approved :: MIT License', - 'Natural Language :: English', - 'Programming Language :: Python :: 3', - 'Programming Language :: Python :: 3.6', - 'Topic :: Software Development :: Libraries :: Python Modules', - 'Topic :: Software Development :: Testing', - 'Topic :: Utilities', - ], - keywords='api recharge', - long_description=long_description, - install_requires=['requests'], - zip_safe=False, - python_requires=">=3.6" -) \ No newline at end of file