Develop

.gitignore

@@ -127,3 +127,5 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+.idea/

README.md

@@ -1,2 +1,3 @@
-# python-substack
-Substack API python implementation
+# Welcome to Python Substack
+
+Updated

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"))

pyproject.toml

@@ -0,0 +1,28 @@
+[tool.poetry]
+name = "python-substack"
+version = "0.0.2"
+description = "A Python wrapper around the Substack API."
+authors = ["Paolo Mazza <mazzapaolo2019@gmail.com>"]
+license = "MIT"
+packages = [
+    { include = "substack" }
+]
+
+readme = "README.md"
+
+repository = "https://github.com/hogier/python-substack"
+homepage = "https://github.com/hogier/python-substack"
+
+keywords = ["substack"]
+
+[tool.poetry.dependencies]
+python = "^3.8"
+python-dotenv = "^0.20.0"
+requests = "^2.28.1"
+
+[tool.poetry.dev-dependencies]
+
+
+[build-system]
+requires = ["poetry-core>=1.0.0"]
+build-backend = "poetry.core.masonry.api"

substack/__init__.py

@@ -0,0 +1,11 @@
+"""A library that provides a Python interface to the Substack API."""
+
+__author__ = "Paolo Mazza"
+__email__ = "mazzapaolo2019@gmail.com"
+__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

@@ -0,0 +1,207 @@
+import logging
+
+import requests
+
+from substack.exceptions import SubstackAPIException, SubstackRequestException
+
+logger = logging.getLogger(__name__)
+
+
+class Api:
+    """
+
+    A python interface into the Substack API
+
+    """
+
+    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:
+            >>> import substack
+            >>> api = substack.Api(email="substack email", password="substack password")
+
+        Args:
+          email:
+          password:
+          base_url:
+            The base URL to use to contact the Substack API.
+            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) -> dict:
+        """
+
+        Args:
+          email:
+          password:
+        """
+
+        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):
+        self._session = requests.Session()
+
+        self.login(email, password)
+
+    @staticmethod
+    def _handle_response(response: requests.Response):
+        """
+
+        Internal helper for handling API responses from the Substack server.
+        Raises the appropriate exceptions when necessary; otherwise, returns the
+        response.
+
+        """
+
+        if not (200 <= response.status_code < 300):
+            raise SubstackAPIException(response.status_code, response.text)
+        try:
+            return response.json()
+        except ValueError:
+            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: draft id
+            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: draft id
+
+        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: draft id
+            send:
+            share_automatically:
+
+        Returns:
+
+        """
+        response = self._session.post(
+            f"{self.publication_url}/drafts/{draft}/publish",
+            json={"send": send, "share_automatically": share_automatically},
+        )
+        return Api._handle_response(response=response)

substack/exceptions.py

@@ -0,0 +1,28 @@
+import json
+
+
+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}"
+        else:
+            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}"
+
+
+class SubstackRequestException(Exception):
+    def __init__(self, message):
+        self.message = message
+
+    def __str__(self):
+        return f"SubstackRequestException: {self.message}"

tests/substack/test_api.py

@@ -0,0 +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):
+    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("")
+        self.assertIsNotNone(posted_draft)