Python库 | pysendgrid-0.26.1.tar.gz

# coding: utf-8 from retry import retry_on_exceptions import requests import datetime import time import json import csv import re class SendGrid(object): def __init__(self, api_user, api_key, url_base="https://sendgrid.com"): self.api_user = api_user self.api_key = api_key self.url_base = url_base self.api_urls = { "newsletter": { # create, clone, edit newsletter "add": "/api/newsletter/add.json", "edit": "/api/newsletter/edit.json", "list": "/api/newsletter/list.json", "get": "/api/newsletter/get.json", "del": "/api/newsletter/delete.json"}, "lists": { # recipient lists "add": "/api/newsletter/lists/add.json", "edit": "/api/newsletter/lists/edit.json", "get": "/api/newsletter/lists/get.json", "del": "/api/newsletter/lists/delete.json"}, "email": { # emails of recipient list "add": "/api/newsletter/lists/email/add.json", "edit": "/api/newsletter/lists/email/edit.json", "get": "/api/newsletter/lists/email/get.json", "del": "/api/newsletter/lists/email/delete.json"}, "recipients": { # recipients for a newsletter "add": "/api/newsletter/recipients/add.json", "get": "/api/newsletter/recipients/get.json", "del": "/api/newsletter/delete.json"}, "schedule": { # schedule to send "add": "/api/newsletter/schedule/add.json", "get": "/api/newsletter/schedule/get.json", "del": "/api/newsletter/schedule/delete.json"}, "identity": { # identities for newsletter "add": "/api/newsletter/identity/add.json", "list": "/api/newsletter/identity/list.json", "get": "/api/newsletter/identity/get.json"}, "subuser": { # create new subuser "add": "/apiv2/customer.add.json", "list": "/apiv2/customer.profile.json", "del": "/apiv2/customer.delete.json", "edit": "/apiv2/customer.profile.json"}, "sendip": { # create new subuser "add": "/apiv2/customer.sendip.json", "get": "/apiv2/customer.ip.json"}, "apps": { "activate": "/apiv2/customer.apps.json", "customize": "/apiv2/customer.apps.json"}, "category": { "create": "/api/newsletter/category/create.json", "del": "/api/newsletter/category/remove.json", "list": "/api/newsletter/category/list.json", "add": "/api/newsletter/category/add.json"}, "stats": { "get": "/apiv2/customer.stats.json"} } def build_params(self, d=None): d = d or {} params = dict(api_user=self.api_user, api_key=self.api_key) params.update(d) return params def build_url(self, api, resource): try: return self.url_base + self.api_urls[api][resource] except KeyError: raise("url not found for %s api and %s resource" % (api, resource)) @retry_on_exceptions(types=[Exception], tries=5, sleep=30) def call(self, api, resource, params=None): url = self.build_url(api, resource) call_params = self.build_params(params or {}) response = requests.post(url, params=call_params) try: response_content = json.loads(response.content) except ValueError: response_content = {'error': re.search(r'<title>([^<]+)</title>', response.content).group(1)} with open("sendgrid.log", "a") as sendgridlog: sendgridlog.write(str(response_content) + " at " + datetime.datetime.now().isoformat() + "\n") return dict(success=True, status_code=response.status_code, url=response.url, response=response_content) def get_newsletter(self, name): return self.call('newsletter', 'get', {"name": name}) def list_newsletter(self, name=None): return self.call('newsletter', 'list', {"name": name} if name else {}) def add_newsletter(self, name, subject, html, text=None, identity=None): if not identity: try: identity = self.list_identity()['response'][0]['identity'] except Exception: raise("You have to inform the identity name") text = text or html # TODO: clean HTML tags d = dict(identity=identity, name=name, subject=subject, text=text, html=html) return self.call('newsletter', 'add', d) def del_newsletter(self, name): return self.call('newsletter', 'del', dict(name=name)) def clone_newsletter(self, existing_name, new_name): existing = self.get_newsletter(existing_name)['response'] if isinstance(existing, dict): new = self.add_newsletter( name=new_name, subject=existing['subject'], html=existing['html'], text=existing['text'], identity=existing['identity'] ) return new return existing def edit_newsletter(self, **fields): '''fields must be name, newname, subject, text, html''' identity = self.list_identity()['response'][0]['identity'] return self.call('newsletter', 'edit', dict(identity=identity, **fields)) def list_identity(self, name=None): return self.call('identity', 'list', {"name": name} if name else {}) def add_identity(self, **fields): '''Fields required are: city, state, zip, address, country, name, email, identity''' return self.call('identity', 'add', dict(**fields)) def get_identity(self, identity): return self.call('identity', 'get', {"identity": identity}) def add_list(self, name): return self.call('lists', 'add', {"list": name}) def get_list(self, name=None): return self.call('lists', 'get', {"list": name} if name else {}) def del_list(self, name): return self.call('lists', 'del', dict(list=name)) def edit_list(self, list, newlist): return self.call('lists', 'edit', dict(list=list, newlist=newlist)) def add_email_to(self, **fields): return self.call('email', 'add', dict(**fields)) def del_email_from(self, **fields): '''fields can be list, and email''' return self.call('email', 'del', dict(**fields)) def add_emails_to(self, list_name, emails): """adds a list of emails [{'email': 'jon@jon.com', 'name': 'Jon'}, {'email': 'mary@mary.com', 'name': 'Mary'}] """ #email_data = [] for i, email in enumerate(emails): emails[i] = json.dumps(email) #data = json.dumps(email_data) return self.call('email', 'add', dict(list=list_name, data=emails)) def get_email(self, list_name, **fields): return self.call('email', 'get', dict(list=list_name, **fields)) def add_recipients(self, newsletter_name, list_name): times = 10 while times: print "trying to add recipient: %s" % times api_call = self.call('recipients', 'add', {"name": newsletter_name, "list": list_name}) if 'error' in api_call['response']: if 'without recipients' in api_call['response']['error']: time.sleep(30) times -= 1 else: times = 0 return api_call def get_recipients(self, newsletter_name): return self.call("recipients", "get", dict(name=newsletter_name)) def del_recipients(self, newsletter_name, list): return self.call("recipients", "de