Skip to content

Instantly share code, notes, and snippets.

@lucassimon
Created January 18, 2017 14:23
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save lucassimon/a23addc04350b260a7f9b8eeedce8e81 to your computer and use it in GitHub Desktop.
Save lucassimon/a23addc04350b260a7f9b8eeedce8e81 to your computer and use it in GitHub Desktop.
metodo de post do formulario. Defini o metada como dicionario
# encoding: utf-8
import requests
from .exceptions import NotBoundException
from .resource import AbstractResource
from .settings import BASE_URL
from .transaction import Transaction
class Plan(AbstractResource):
BASE_URL = BASE_URL + 'plans'
def __init__(self, api_key='', name='', amount=None, days=None, installments=1, payment_methods=['boleto', 'credit_card'],
color=None, charges=1, trial_days=0, **kwargs):
if not api_key:
raise ValueError('You should supply an api_key')
for payment_method in payment_methods:
if payment_method not in ['boleto', 'credit_card']:
raise ValueError('Invalid payment method, try a list with "boleto" and/or "credit_card"')
self.data = {'api_key': api_key}
self.data['name'] = name
self.data['amount'] = amount
self.data['installments'] = installments
self.data['payment_methods'] = payment_methods
self.data['color'] = color
self.data['charges'] = charges
self.data['trial_days'] = trial_days
self.data['days'] = days
self.data.update(kwargs)
def find_by_id(self, id):
url = self.BASE_URL + '/' + str(id)
data = {'api_key': self.data['api_key']}
pagarme_response = requests.get(url, params=data)
if pagarme_response.status_code == 200:
self.handle_response(pagarme_response.json())
else:
self.error(pagarme_response.json())
class Subscription(AbstractResource):
BASE_URL = BASE_URL + 'subscriptions'
def __init__(
self,
api_key=None,
plan_id=None,
card_id=None,
card_hash=None,
postback_url=None,
metadata={},
customer=None,
**kwargs):
if not api_key:
raise ValueError('Invalid api_key')
if plan_id and not isinstance(plan_id, int):
raise ValueError('plan_id should be a int')
if customer and not customer.data['email']:
raise ValueError('Customer email not found')
self.data = {
'api_key': api_key,
'plan_id': plan_id,
'card_hash': card_hash,
'postback_url': postback_url,
}
self.metadata = metadata
self.customer = customer
if card_hash:
self.data['card_hash'] = card_hash
else:
self.data['card_id'] = card_id
self.data.update(kwargs)
def get_data(self):
data = self.data
if self.customer:
data.update(self.customer.get_anti_fraud_data())
if self.metadata:
for key, value in self.metadata.items():
new_key = 'metadata[{key}]'.format(key=key)
data[new_key] = value
return data
def find_by_id(self, id):
url = self.BASE_URL + '/' + str(id)
data = {'api_key': self.data['api_key']}
pagarme_response = requests.get(url, params=data)
if pagarme_response.status_code == 200:
self.handle_response(pagarme_response.json())
else:
self.error(pagarme_response.json())
def cancel(self):
if not self.data.get('id', False):
raise NotBoundException('First try search your subscription')
url = self.BASE_URL + '/{id}/cancel'.format(id=self.data['id'])
pagarme_response = requests.post(url, data={'api_key': self.data['api_key']})
if pagarme_response.status_code == 200:
self.handle_response(pagarme_response.json())
else:
self.error(pagarme_response.json())
def transactions(self):
if not self.data.get('id', False):
raise NotBoundException('First try search your subscription')
url = self.BASE_URL + '/{id}/transactions'.format(id=self.data['id'])
pagarme_response = requests.get(url, params={'api_key': self.data['api_key']})
if pagarme_response.status_code != 200:
self.error(pagarme_response.json())
response = pagarme_response.json()
transactions = []
for transaction in response:
t = Transaction(self.data['api_key'])
t.handle_response(transaction)
transactions.append(t)
return transactions
# encoding: utf-8
# ====================
#VERSAO DO PYPI sem referencia ao metadata
# ====================# encoding: utf-8
import json
import requests
from .exceptions import NotBoundException
from .resource import AbstractResource
from .settings import BASE_URL
from .transaction import Transaction
# encoding: utf-8
import json
import requests
from .exceptions import PagarmeApiError
class AbstractResource(object):
def __init__(self):
raise NotImplementedError
def handle_response(self, data):
self.data.update(data)
def error(self, response):
data = json.loads(response)
e = data['errors'][0]
error_string = e['type'] + ' - ' + e['message']
raise PagarmeApiError(error_string)
def create(self):
url = self.BASE_URL
pagarme_response = requests.post(url, data=self.get_data())
if pagarme_response.status_code == 200:
self.handle_response(json.loads(pagarme_response.content))
else:
self.error(pagarme_response.content)
def get_data(self):
return self.data
class Plan(AbstractResource):
BASE_URL = BASE_URL + 'plans'
def __init__(self, api_key='', name='', amount=None, days=None, installments=1, payment_methods=['boleto', 'credit_card'],
color=None, charges=1, trial_days=0, **kwargs):
if not api_key:
raise ValueError('You should supply an api_key')
for payment_method in payment_methods:
if payment_method not in ['boleto', 'credit_card']:
raise ValueError('Invalid payment method, try a list with "boleto" and/or "credit_card"')
self.data = {'api_key': api_key}
self.data['name'] = name
self.data['amount'] = amount
self.data['installments'] = installments
self.data['payment_methods'] = payment_methods
self.data['color'] = color
self.data['charges'] = charges
self.data['trial_days'] = trial_days
self.data['days'] = days
self.data.update(kwargs)
def find_by_id(self, id):
url = self.BASE_URL + '/' + str(id)
data = {'api_key': self.data['api_key']}
pagarme_response = requests.get(url, params=data)
if pagarme_response.status_code == 200:
self.handle_response(json.loads(pagarme_response.content))
else:
self.error(pagarme_response.content)
class Subscription(AbstractResource):
BASE_URL = BASE_URL + 'subscriptions'
def __init__(
self,
api_key=None,
plan_id=None,
card_id=None,
card_hash=None,
postback_url=None,
customer=None,
**kwargs):
if not api_key:
raise ValueError('Invalid api_key')
if plan_id and not isinstance(plan_id, int):
raise ValueError('plan_id should be a int')
if customer and not customer.data['email']:
raise ValueError('Customer email not found')
self.data = {
'api_key': api_key,
'plan_id': plan_id,
'card_hash': card_hash,
'postback_url': postback_url,
}
self.customer = customer
if card_hash:
self.data['card_hash'] = card_hash
else:
self.data['card_id'] = card_id
self.data.update(kwargs)
def get_data(self):
data = self.data
if self.customer:
data.update(self.customer.get_anti_fraud_data())
return data
def find_by_id(self, id):
url = self.BASE_URL + '/' + str(id)
data = {'api_key': self.data['api_key']}
pagarme_response = requests.get(url, params=data)
if pagarme_response.status_code == 200:
self.handle_response(json.loads(pagarme_response.content))
else:
self.error(pagarme_response.content)
def cancel(self):
if not self.data.get('id', False):
raise NotBoundException('First try search your subscription')
url = self.BASE_URL + '/{id}/cancel'.format(id=self.data['id'])
pagarme_response = requests.post(url, data={'api_key': self.data['api_key']})
if pagarme_response.status_code == 200:
self.handle_response(json.loads(pagarme_response.content))
else:
self.error(pagarme_response.content)
def transactions(self):
if not self.data.get('id', False):
raise NotBoundException('First try search your subscription')
url = self.BASE_URL + '/{id}/transactions'.format(id=self.data['id'])
pagarme_response = requests.get(url, params={'api_key': self.data['api_key']})
if pagarme_response.status_code != 200:
self.error(pagarme_response.content)
response = json.loads(pagarme_response.content)
transactions = []
for transaction in response:
t = Transaction(self.data['api_key'])
t.handle_response(transaction)
transactions.append(t)
return transactions
def post(self, request, *args, **kwargs):
company = Company.objects.get(
userprofile=self.request.user.get_profile()
)
plan = None
try:
plan = Plan.objects.get(pk=request.POST.get('plan', None))
except Plan.DoesNotExist:
messages.error(
self.request,
'Plano não existe.'
)
return redirect(
reverse_lazy(
'subscriptions:add-subscription'
)
)
subscription_form = SubscriptionForm(request.POST)
if subscription_form.is_valid():
subscription = subscription_form.save(commit=False)
subscription.company = company
card_hash = request.POST.get('card_hash', None)
# salvar dados do cardhash
# subscription.card_hash = card_hash
# FAZER REQUEST PARA A API e pegar o id da subscription gerada
pagarme_metadata = {
'company_id': company.pk,
'user_id': self.request.user.pk
}
customer = PCustomer(
name=company.company_name,
document_number=company.br_cnpj,
email=company.email,
)
try:
p_subscription = PSubscription(
api_key=settings.PAGARME_API_KEY,
plan_id=int(plan.pagarme_plan_id),
card_hash=card_hash,
customer=customer,
metadata=pagarme_metadata
)
except Exception:
messages.error(
request,
(
'Ocorreu um erro ao salvar a assinatura na API. '
'Contate o administrador do sistema.'
)
)
return redirect(
reverse_lazy(
'subscriptions:add-subscription'
)
)
try:
p_subscription.create()
except Exception:
messages.error(
request,
(
'Ocorreu um erro ao salvar a assinatura na API. '
'Contate o administrador do sistema.'
)
)
client.captureException()
data = p_subscription.data
subscription.pagarme_subscription_id = data['id']
current_period_start = data.get(
'current_period_start'
).replace(' ', '')
current_period_start = current_period_start.split('T')[0]
subscription.pagarme_current_period_date = (
datetime.datetime.strptime(
current_period_start.split('T')[0], '%Y-%m-%d'
)
)
subscription.save()
messages.success(
request,
'Assinatura criada com sucesso'
)
return redirect(
reverse_lazy(
'subscriptions:subscription'
)
)
else:
subscription_form.fields["plan"].queryset = (
Plan.objects.all()
)
context = {
'title': 'Assinatura',
'description': 'Adicionar uma assinatura',
'subscription_form': subscription_form,
}
messages.error(
request,
(
'Ocorreu um erro ao salvar o consulta. '
'Confira os formulários'
)
)
return render(request, self.template_name, context)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment