first features

examples/publish_post.py

@@ -0,0 +1,30 @@
+import os
+
+from dotenv import load_dotenv
+
+from substack import Api
+
+load_dotenv()
+
+content = ""
+title = ""
+subtitle = ""
+
+api = Api(
+    email=os.getenv("EMAIL"),
+    password=os.getenv("PASSWORD"),
+    publication_url=os.getenv("PUBLICATION_URL"),
+)
+
+body = f'{{"type":"doc","content": {content}}}'
+
+draft = api.post_draft(
+    [{"id": os.getenv("USER_ID"), "is_guest": False}],
+    title=title,
+    subtitle=subtitle,
+    body=body,
+)
+
+api.prepublish_draft(draft.get("id"))
+
+api.publish_draft(draft.get("id"))

substack/__init__.py

@@ -1,11 +1,11 @@
 """A library that provides a Python interface to the Substack API."""
 
-__author__ = 'Paolo Mazza'
-__email__ = '[email protected]'
-__license__ = 'MIT License'
-__version__ = '1.0'
-__url__ = 'https://github.com/hogier/python-substack'
-__download_url__ = 'https://pypi.python.org/pypi/python-substack'
-__description__ = 'A Python wrapper around the Substack API'
+__author__ = "Paolo Mazza"
+__email__ = "[email protected]"
+__license__ = "MIT License"
+__version__ = "1.0"
+__url__ = "https://github.com/hogier/python-substack"
+__download_url__ = "https://pypi.python.org/pypi/python-substack"
+__description__ = "A Python wrapper around the Substack API"
 
 from .api import Api

substack/api.py

@@ -14,7 +14,14 @@ class Api:
 
     """
 
-    def __init__(self, email: str, password: str, base_url: str | None = None, debug: bool = False):
+    def __init__(
+        self,
+        email: str,
+        password: str,
+        base_url: str | None = None,
+        publication_url: str | None = None,
+        debug: bool = False,
+    ):
         """
 
         To create an instance of the substack.Api class:
@@ -29,26 +36,32 @@ def __init__(self, email: str, password: str, base_url: str | None = None, debug
             Defaults to https://substack.com/api/v1.
         """
         self.base_url = base_url or "https://substack.com/api/v1"
+        self.publication_url = publication_url
 
         if debug:
             logging.basicConfig()
             logging.getLogger().setLevel(logging.DEBUG)
 
         self._init_session(email, password)
 
-    def login(self, email: str, password: str):
+    def login(self, email: str, password: str) -> dict:
         """
 
         Args:
           email:
           password:
         """
 
-        response = self._session.post(f"{self.base_url}/login", json={"captcha_response": None,
-                                                                      "email": email,
-                                                                      "for_pub": "",
-                                                                      "password": password,
-                                                                      "redirect": "/"})
+        response = self._session.post(
+            f"{self.base_url}/login",
+            json={
+                "captcha_response": None,
+                "email": email,
+                "for_pub": "",
+                "password": password,
+                "redirect": "/",
+            },
+        )
         return Api._handle_response(response=response)
 
     def _init_session(self, email, password):
@@ -71,4 +84,124 @@ def _handle_response(response: requests.Response):
         try:
             return response.json()
         except ValueError:
-            raise SubstackRequestException('Invalid Response: %s' % response.text)
+            raise SubstackRequestException("Invalid Response: %s" % response.text)
+
+    def get_publication_users(self):
+        """
+
+        :return:
+        """
+        response = self._session.get(f"{self.publication_url}/publication/users")
+
+        return Api._handle_response(response=response)
+
+    def get_posts(self) -> dict:
+        """
+
+        :return:
+        """
+        response = self._session.get(f"{self.base_url}/reader/posts")
+
+        return Api._handle_response(response=response)
+
+    def get_drafts(self, filter: str = None, offset: int = None, limit: int = None):
+        response = self._session.get(
+            f"{self.publication_url}/drafts",
+            params={"filter": filter, "offset": offset, "limit": limit},
+        )
+        return Api._handle_response(response=response)
+
+    def post_draft(
+        self,
+        draft_bylines: list,
+        title: str = None,
+        subtitle: str = None,
+        body: str = None,
+    ) -> dict:
+        """
+
+        Args:
+          draft_bylines:
+          title:
+          subtitle:
+          body:
+
+        Returns:
+
+        """
+        response = self._session.post(
+            f"{self.publication_url}/drafts",
+            json={
+                "draft_bylines": draft_bylines,
+                "draft_title": title,
+                "draft_subtitle": subtitle,
+                "draft_body": body,
+            },
+        )
+        return Api._handle_response(response=response)
+
+    def put_draft(
+        self,
+        draft: str,
+        title: str = None,
+        subtitle: str = None,
+        body: str = None,
+        cover_image: str = None,
+    ) -> dict:
+        """
+
+        Args:
+            draft:
+            title:
+            subtitle:
+            body:
+            cover_image:
+
+        Returns:
+
+        """
+
+        response = self._session.put(
+            f"{self.publication_url}/drafts/{draft}",
+            json={
+                "draft_title": title,
+                "draft_subtitle": subtitle,
+                "draft_body": body,
+                "cover_image": cover_image,
+            },
+        )
+        return Api._handle_response(response=response)
+
+    def prepublish_draft(self, draft: str) -> dict:
+        """
+
+        Args:
+            draft:
+
+        Returns:
+
+        """
+
+        response = self._session.get(
+            f"{self.publication_url}/drafts/{draft}/prepublish"
+        )
+        return Api._handle_response(response=response)
+
+    def publish_draft(
+        self, draft: str, send: bool = True, share_automatically: bool = False
+    ) -> dict:
+        """
+
+        Args:
+            draft:
+            send:
+            share_automatically:
+
+        Returns:
+
+        """
+        response = requests.post(
+            f"{self.publication_url}/drafts/{draft}/publish",
+            json={"send": send, "share_automatically": share_automatically},
+        )
+        return Api._handle_response(response=response)

substack/exceptions.py

@@ -2,24 +2,27 @@
 
 
 class SubstackAPIException(Exception):
-
     def __init__(self, status_code, text):
         try:
             json_res = json.loads(text)
         except ValueError:
-            self.message = f'Invalid JSON error message from Substack: {text}'
+            self.message = f"Invalid JSON error message from Substack: {text}"
         else:
-            self.message = ", ".join(list(map(lambda error: error.get("msg", ""), json_res.get("errors", []))))
+            self.message = ", ".join(
+                list(
+                    map(lambda error: error.get("msg", ""), json_res.get("errors", []))
+                )
+            )
             self.message = self.message or json_res.get("error", "")
         self.status_code = status_code
 
     def __str__(self):
-        return f'APIError(code={self.status_code}): {self.message}'
+        return f"APIError(code={self.status_code}): {self.message}"
 
 
 class SubstackRequestException(Exception):
     def __init__(self, message):
         self.message = message
 
     def __str__(self):
-        return f'SubstackRequestException: {self.message}'
+        return f"SubstackRequestException: {self.message}"

tests/substack/test_api.py

@@ -1,11 +1,64 @@
+import os
 import unittest
 
+from dotenv import load_dotenv
+
 from substack import Api
 from substack.exceptions import SubstackAPIException
 
+load_dotenv()
 
-class ApiTest(unittest.TestCase):
 
+class ApiTest(unittest.TestCase):
     def test_api_exception(self):
         with self.assertRaises(SubstackAPIException):
             Api(email="", password="")
+
+    def test_login(self):
+        api = Api(
+            email=os.getenv("EMAIL"),
+            password=os.getenv("PASSWORD"),
+            publication_url=os.getenv("PUBLICATION_URL"),
+        )
+        self.assertIsNotNone(api)
+
+    def test_get_posts(self):
+        api = Api(email=os.getenv("EMAIL"), password=os.getenv("PASSWORD"))
+        posts = api.get_posts()
+        self.assertIsNotNone(posts)
+
+    def test_get_drafts(self):
+        api = Api(
+            email=os.getenv("EMAIL"),
+            password=os.getenv("PASSWORD"),
+            publication_url=os.getenv("PUBLICATION_URL"),
+        )
+        drafts = api.get_drafts()
+        self.assertIsNotNone(drafts)
+
+    def test_post_draft(self):
+        api = Api(
+            email=os.getenv("EMAIL"),
+            password=os.getenv("PASSWORD"),
+            publication_url=os.getenv("PUBLICATION_URL"),
+        )
+        posted_draft = api.post_draft([{"id": os.getenv("USER_ID"), "is_guest": False}])
+        self.assertIsNotNone(posted_draft)
+
+    def test_publication_users(self):
+        api = Api(
+            email=os.getenv("EMAIL"),
+            password=os.getenv("PASSWORD"),
+            publication_url=os.getenv("PUBLICATION_URL"),
+        )
+        users = api.get_publication_users()
+        self.assertIsNotNone(users)
+
+    def test_put_draft(self):
+        api = Api(
+            email=os.getenv("EMAIL"),
+            password=os.getenv("PASSWORD"),
+            publication_url=os.getenv("PUBLICATION_URL"),
+        )
+        posted_draft = api.put_draft("62667935")
+        self.assertIsNotNone(posted_draft)