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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 51 additions & 46 deletions barte/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,8 @@ def __init__(self, api_key: str, environment: str = "production"):
else "https://sandbox-api.barte.com"
)
self.headers = {"X-Token-Api": api_key, "Content-Type": "application/json"}
self.session = requests.Session()
self.session.headers.update(self.headers)
BarteClient._instance = self

@classmethod
Expand All @@ -53,76 +55,79 @@ def get_instance(cls) -> "BarteClient":
)
return cls._instance

def _request(
self,
method: str,
path: str,
params: Optional[Dict[str, Any]] = None,
json: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""
Private method to centralize HTTP requests.

Args:
method: HTTP method (e.g., 'GET', 'POST', 'DELETE', etc.)
path: API endpoint path (e.g., '/v2/orders')
params: Query parameters for GET requests.
json: JSON body for POST, PATCH requests.

Returns:
The response JSON as a dictionary.

Raises:
HTTPError: If the HTTP request returned an unsuccessful status code.
"""
url = f"{self.base_url}{path}"
response = self.session.request(method, url, params=params, json=json)
response.raise_for_status()
return response.json()

def create_order(self, data: Union[Dict[str, Any], OrderPayload]) -> Order:
"""Create a new order"""
endpoint = f"{self.base_url}/v2/orders"

if isinstance(data, OrderPayload):
data = asdict(data)

response = requests.post(endpoint, headers=self.headers, json=data)
response.raise_for_status()
return from_dict(data_class=Order, data=response.json(), config=DACITE_CONFIG)
json_response = self._request("POST", "/v2/orders", json=data)
return from_dict(data_class=Order, data=json_response, config=DACITE_CONFIG)

def get_charge(self, charge_id: str) -> Charge:
"""Get a specific charge"""
endpoint = f"{self.base_url}/v2/charges/{charge_id}"
response = requests.get(endpoint, headers=self.headers)
response.raise_for_status()
return from_dict(data_class=Charge, data=response.json(), config=DACITE_CONFIG)
json_response = self._request("GET", f"/v2/charges/{charge_id}")
return from_dict(data_class=Charge, data=json_response, config=DACITE_CONFIG)

def list_charges(self, params: Optional[Dict[str, Any]] = None) -> ChargeList:
"""List all charges with optional filters"""
endpoint = f"{self.base_url}/v2/charges"
response = requests.get(endpoint, headers=self.headers, params=params)
response.raise_for_status()
json_response = self._request("GET", "/v2/charges", params=params)
return from_dict(
data_class=ChargeList, data=response.json(), config=DACITE_CONFIG
data_class=ChargeList, data=json_response, config=DACITE_CONFIG
)

def cancel_charge(self, charge_id: str) -> None:
"""Cancel a specific charge"""
endpoint = f"{self.base_url}/v2/charges/{charge_id}"
response = requests.delete(endpoint, headers=self.headers)
response.raise_for_status()
self._request("DELETE", f"/v2/charges/{charge_id}")

def create_buyer(self, buyer_data: Dict[str, any]) -> Buyer:
endpoint = f"{self.base_url}/v2/buyers"
response = requests.post(endpoint, headers=self.headers, json=buyer_data)
response.raise_for_status()
return from_dict(data_class=Buyer, data=response.json(), config=DACITE_CONFIG)
def create_buyer(self, buyer_data: Dict[str, Any]) -> Buyer:
"""Create a buyer"""
json_response = self._request("POST", "/v2/buyers", json=buyer_data)
return from_dict(data_class=Buyer, data=json_response, config=DACITE_CONFIG)

def get_buyer(self, filters: Dict[str, any]) -> BuyerList:
endpoint = f"{self.base_url}/v2/buyers"
response = requests.get(endpoint, params=filters, headers=self.headers)
response.raise_for_status()
return from_dict(
data_class=BuyerList, data=response.json(), config=DACITE_CONFIG
)
def get_buyer(self, filters: Dict[str, Any]) -> BuyerList:
"""Get buyers based on filters"""
json_response = self._request("GET", "/v2/buyers", params=filters)
return from_dict(data_class=BuyerList, data=json_response, config=DACITE_CONFIG)

def create_card_token(self, card_data: Dict[str, Any]) -> CardToken:
"""Create a token for a credit card"""
endpoint = f"{self.base_url}/v2/cards"
response = requests.post(endpoint, headers=self.headers, json=card_data)
response.raise_for_status()
return from_dict(
data_class=CardToken, data=response.json(), config=DACITE_CONFIG
)
json_response = self._request("POST", "/v2/cards", json=card_data)
return from_dict(data_class=CardToken, data=json_response, config=DACITE_CONFIG)

def get_pix_qrcode(self, charge_id: str) -> PixCharge:
"""Get PIX QR Code data for a charge"""
endpoint = f"{self.base_url}/v2/charges/{charge_id}"
response = requests.get(endpoint, headers=self.headers)
response.raise_for_status()
return from_dict(
data_class=PixCharge, data=response.json(), config=DACITE_CONFIG
)
json_response = self._request("GET", f"/v2/charges/{charge_id}")
return from_dict(data_class=PixCharge, data=json_response, config=DACITE_CONFIG)

def refund_charge(self, charge_id: str, as_fraud: Optional[bool] = False) -> Refund:
"""Refund a charge"""
endpoint = f"{self.base_url}/v2/charges/{charge_id}/refund"
response = requests.patch(
endpoint, headers=self.headers, json={"asFraud": as_fraud}
json_response = self._request(
"PATCH", f"/v2/charges/{charge_id}/refund", json={"asFraud": as_fraud}
)
response.raise_for_status()
return from_dict(data_class=Refund, data=response.json(), config=DACITE_CONFIG)
return from_dict(data_class=Refund, data=json_response, config=DACITE_CONFIG)
Loading