sendgrid · kylesalk · Oct 7, 2018 · Oct 7, 2018 · Oct 7, 2018 · Oct 7, 2018
diff --git a/examples/helpers/campaigns/campaigns.py b/examples/helpers/campaigns/campaigns.py
@@ -0,0 +1,76 @@
+import json
+import os
+from sendgrid.helpers.campaigns import *
+from sendgrid import *
+
+# NOTE: you will need move this file to the root directory of this project to execute properly.
+
+# Assumes you set your environment variable:
+# https://github.com/sendgrid/sendgrid-python/blob/master/TROUBLESHOOTING.md#environment-variables-and-your-sendgrid-api-key
+SG = SendGridAPIClient(apikey=os.environ.get('SENDGRID_API_KEY'))
+
+
+def pprint_json(json_raw):
+    print(json.dumps(json.loads(json_raw), indent=4, sort_keys=True))
+
+
+def print_campaigns():
+    response = sg.client.categories.campaigns.get(query_params=stats_params)
+    print(response.status_code)
+    print(response.headers)
+    pprint_json(response.body)
+
+
+def get_campaigns():
+    """Get Campaign objects"""
+    campaigns = []
+    response = SG.client.campaigns.get()
+    for camp in json.loads(response.body.decode())["result"]:
+        campaigns.append(Campaign(**camp))
+    return campaigns
+
+
+def create_campaign(call_func=False):
+    camp_settings = {
+        "title": "Test Campaign 1",
+        "categories": ["test", "example"],
+        "html_content": "<html><head><title>New Edition!</title></head><body><p>New edition is now live!</p></body></html>",
+    }
+    camp = Campaign(**camp_settings)
+
+    if call_func:
+        campaign_build(SG, camp)
+    else:
+        SG.client.campaigns.post(request_body=camp.get())
+
+
+def schedule_campaign():
+    """Schedule the first campaign retrieved"""
+    # campaigns = []
+
+    campaigns = Campaigns(offset=0, limit=2)
+    response = SG.client.campaigns.get(query_params=campaigns.get())
+    for camp in json.loads(response.body.decode())["result"]:
+        print(schedule.get())
+        print(camp["title"])
+        c_response = getattr(SG.client.campaigns, str(camp["id"])).schedules.get()
+        pprint_json(c_response.body.decode())
+
+        schedule = Schedule(year=2018, month=12, day=1, hour=8, minute=23)
+        c_response = getattr(SG.client.campaigns, str(camp["id"])).schedules.post(
+            request_body=schedule.get()
+        )
+        pprint_json(c_response.body.decode())
+
+        c_response = getattr(SG.client.campaigns, str(camp["id"])).schedules.get()
+        pprint_json(c_response.body.decode())
+        break
+
+
+def main():
+    get_campaigns()
+    create_campaign()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/sendgrid/helpers/campaigns/__init__.py b/sendgrid/helpers/campaigns/__init__.py
@@ -0,0 +1,6 @@
+from .campaign import Campaign
+from .campaigns import Campaigns
+from .schedule import Schedule
+from .campaign_build import campaign_build
+from .campaign_build_send import campaign_build_send
+from .campaign_build_scheduled import campaign_build_scheduled
diff --git a/sendgrid/helpers/campaigns/campaign.py b/sendgrid/helpers/campaigns/campaign.py
@@ -0,0 +1,230 @@
+class Campaign(object):
+    """Campaign class object for campaign API queries
+
+    Required parameter:
+        title: str
+
+    Optional parameters:
+        categories: list(str)
+        custom_unsubscribe_url: str
+        editor: str ['code' | 'design']
+        html_content: str
+        ip_pool: str
+        list_ids: list(int)
+        plain_content: str
+        segment_ids: list(int)
+        sender_id: int
+        subject: str
+        suppression_group_id: int
+
+    :param kwargs: List of inputs
+    """
+    def __init__(self, **kwargs):
+        self._categories = []
+        self._custom_unsubscribe_url = None
+        self._editor = None
+        self._html_content = None
+        self._id = None
+        self._ip_pool = None
+        self._list_ids = []
+        self._plain_content = None
+        self._segment_ids = []
+        self._sender_id = None
+        self._status = ""
+        self._subject = None
+        self._suppression_group_id = None
+        self._title = ""
+
+        for key, val in kwargs.items():
+            if hasattr(self, key):
+                setattr(self, key, val)
+
+    def copy(self, new_title):
+        """Creates a copy of the campaign
+
+        :param new_title: Title of new campaign
+        :type new_title: str
+        :return: Campaign
+        """
+        new_camp = Campaign(**self.get())
+        new_camp.title = new_title
+        return new_camp
+
+    def get(self):
+        """Returns dict suitable for queries"""
+        body = {"title": self.title}
+        if self.categories:
+            body["categories"] = [str(i) for i in self.categories]
+        if self.custom_unsubscribe_url is not None:
+            body["custom_unsubscribe_url"] = self.custom_unsubscribe_url
+        if self.editor is not None:
+            body["editor"] = self.editor
+        if self.html_content is not None:
+            body["html_content"] = self.html_content
+        if self.ip_pool is not None:
+            body["ip_pool"] = self.ip_pool
+        if self.list_ids and all(isinstance(i, int) for i in self.list_ids):
+            body["list_ids"] = self.list_ids
+        if self.plain_content:
+            body["plain_content"] = self.plain_content
+        if self.segment_ids\
+                and all(isinstance(i, int) for i in self.segment_ids):
+            body["segment_ids"] = self.segment_ids
+        if self.sender_id is not None:
+            body["sender_id"] = self.sender_id
+        if self.subject is not None:
+            body["subject"] = self.subject
+        if self.suppression_group_id is not None:
+            body["suppression_group_id"] = self.suppression_group_id
+        return body
+
+    def get_patch(self):
+        return {
+            "title": self.title,
+            "subject": self.subject,
+            "categories": self.categories,
+            "html_content": self.html_content,
+            "plain_content": self.plain_content
+        }
+
+    def patch(self, **kwargs):
+        """Updates the campaign with
+
+        Optional inputs:
+            title: str
+            subject: str
+            categories: list(str)
+            html_content: str
+            plain_content: str
+
+        :param kwargs: Dictionary of optional inputs
+        :type kwargs: dict
+        :return: Updated get() dict
+        """
+        if "title" in kwargs:
+            self.title = kwargs["title"]
+        if "subject" in kwargs:
+            self.subject = kwargs["subject"]
+        if "categories" in kwargs:
+            self.categories = kwargs["categories"]
+        if "html_content" in kwargs:
+            self.html_content = kwargs["html_content"]
+        if "plain_content" in kwargs:
+            self.plain_content = kwargs["plain_content"]
+        return self.get_patch()
+
+    @property
+    def categories(self):
+        return self._categories
+
+    @categories.setter
+    def categories(self, value):
+        self._categories = value
+
+    @property
+    def custom_unsubscribe_url(self):
+        return self._custom_unsubscribe_url
+
+    @custom_unsubscribe_url.setter
+    def custom_unsubscribe_url(self, value):
+        self._custom_unsubscribe_url = value
+
+    @property
+    def editor(self):
+        return self._editor
+
+    @editor.setter
+    def editor(self, value):
+        value = str(value).lower()
+        if value in ["code", "design"]:
+            self._editor = value
+
+    @property
+    def html_content(self):
+        return self._html_content
+
+    @html_content.setter
+    def html_content(self, value):
+        self._html_content = value
+
+    @property
+    def id(self):
+        return self._id
+
+    @id.setter
+    def id(self, value):
+        self._id = value
+
+    @property
+    def ip_pool(self):
+        return self._ip_pool
+
+    @ip_pool.setter
+    def ip_pool(self, value):
+        self._ip_pool = value
+
+    @property
+    def list_ids(self):
+        return self._list_ids
+
+    @list_ids.setter
+    def list_ids(self, value):
+        self._list_ids = value
+
+    @property
+    def plain_content(self):
+        return self._plain_content
+
+    @plain_content.setter
+    def plain_content(self, value):
+        self._plain_content = value
+
+    @property
+    def segment_ids(self):
+        return self._segment_ids
+
+    @segment_ids.setter
+    def segment_ids(self, value):
+        self._segment_ids = value
+
+    @property
+    def sender_id(self):
+        return self._sender_id
+
+    @sender_id.setter
+    def sender_id(self, value):
+        self._sender_id = value
+
+    @property
+    def status(self):
+        return self._status
+
+    @status.setter
+    def status(self, value):
+        self._status = value
+
+    @property
+    def subject(self):
+        return self._subject
+
+    @subject.setter
+    def subject(self, value):
+        if isinstance(value, str) or isinstance(value, type(None)):
+            self._subject = value
+
+    @property
+    def suppression_group_id(self):
+        return self._suppression_group_id
+
+    @suppression_group_id.setter
+    def suppression_group_id(self, value):
+        self._suppression_group_id = value
+
+    @property
+    def title(self):
+        return self._title
+
+    @title.setter
+    def title(self, value):
+        if value:
+            self._title = str(value)
diff --git a/sendgrid/helpers/campaigns/campaign_build.py b/sendgrid/helpers/campaigns/campaign_build.py
@@ -0,0 +1,24 @@
+import json
+
+
+def campaign_build(api_client, campaign, notify=None):
+    """Build a Campaign via the SendGrid API
+
+    :param api_client: SendGrid API Client
+    :type api_client: sendgrid.SendGridAPIClient
+    :param campaign: Campaign object to build
+    :type campaign: sendgrid.helpers.campaigns.Campaign
+    :param notify: Mail objects to send on completion
+    :type notify: list,sendgrid.helpers.mail.mail.Mail
+    :return: Campaign.id for newly created Campaign
+    :rtype: int
+    """
+    response = api_client.client.campaigns.post(request_body=campaign.get())
+    camp_id = json.loads(response.body.decode())["id"]
+    if notify is not None:
+        if isinstance(notify, list):
+            for n in notify:
+                api_client.client.mail.send.post(request_body=n.get())
+        else:
+            api_client.client.mail.send.post(request_body=notify.get())
+    return camp_id
diff --git a/sendgrid/helpers/campaigns/campaign_build_scheduled.py b/sendgrid/helpers/campaigns/campaign_build_scheduled.py
@@ -0,0 +1,30 @@
+from .campaign_build import campaign_build
+
+
+def campaign_build_scheduled(api_client, campaign, schedule, notify=None):
+    """Builds a campaign and schedules for release
+
+    See sendgrid.helpers.campaigns.campaign.get_schedule for kwargs
+
+    :param api_client: SendGrid API Client
+    :type api_client: sendgrid.SendGridAPIClient
+    :param campaign: Campaign object to build
+    :type campaign: sendgrid.helpers.campaigns.Campaign
+    :param notify: Mail objects to send on completion
+    :type notify: list,sendgrid.helpers.mail.mail.Mail
+    :param schedule: Schedule object to get schedule body
+    :type schedule: sendgrid.helpers.campaigns.schedule.Schedule
+    :return: ID of created campaign
+    :rtype: int
+    """
+    c_id = campaign_build(api_client, campaign)
+    getattr(api_client.client.campaigns, str(c_id)).schedules.post(
+        request_body=schedule.get()
+    )
+    if notify is not None:
+        if isinstance(notify, list):
+            for n in notify:
+                api_client.client.mail.send.post(request_body=n.get())
+        else:
+            api_client.client.mail.send.post(request_body=notify.get())
+    return c_id
diff --git a/sendgrid/helpers/campaigns/campaign_build_send.py b/sendgrid/helpers/campaigns/campaign_build_send.py
@@ -0,0 +1,24 @@
+from .campaign_build import campaign_build
+
+
+def campaign_build_send(api_client, campaign, notify=None):
+    """Builds a campaign and sends immediately
+
+    :param api_client: SendGrid API Client
+    :type api_client: sendgrid.SendGridAPIClient
+    :param campaign: Campaign object to build
+    :type campaign: sendgrid.helpers.campaigns.Campaign
+    :param notify: Mail objects to send on completion
+    :type notify: list,sendgrid.helpers.mail.mail.Mail
+    :return: ID of created campaign
+    :rtype: int
+    """
+    c_id = campaign_build(api_client, campaign)
+    getattr(api_client.client.campaigns, str(c_id)).schedules.now.post()
+    if notify is not None:
+        if isinstance(notify, list):
+            for n in notify:
+                api_client.client.mail.send.post(request_body=n.get())
+        else:
+            api_client.client.mail.send.post(request_body=notify.get())
+    return c_id