v0.2.0 bump, ready for pypi

Author: AbstractUmbra

.gitignore

@@ -1,9 +1,7 @@
 .vscode/
 __ignore__/
-.idea/
-*.pyc
 __pycache__/
+*.pyc
 /dist/
 /*.egg-info/
-.mypy_cache/
-setup.py
+.mypy_cache/

README.md

@@ -5,13 +5,55 @@ A small simple wrapper around the [Mystb.in](https://mystb.in/) API.
 ### Features
 
 [x] - `POST`ing to the API, which will return the provided url. \
-[x] - `GET`ting from the API, provided you know the URL or paste ID. \
-[ ] - Ability to pass in a sync or async session / parameter so it is flexible.
+[ ] - `GET`ting from the API, provided you know the URL or paste ID. \
+[x] - Ability to pass in a sync or async session / parameter so it is flexible.
 
-[ ] - Write a real underlying Client for this, it will be required for... \
+[x] - Write a real underlying Client for this, it will be required for... \
 [ ] - Authorization. Awaiting the API making this public as it is still WIP. \
 
+### Installation
+This project will be on [PyPI](https://pypi.org/project/mystbin.py/) as a stable release, you can always find that there.
+
+Installing via `pip`:
+```shell
+python -m pip install -U mystbin.py
+# or for optional sync addon...
+python -m pip install -U mystbin.py[requests]
+```
+
+Installing from source:
+```shell
+python -m pip install git+https://github.com/AbstractUmbra/mystbin-py.git #[requests] for sync addon
+```
+
+### Usage examples
+Since the project is considered multi-sync, it will work in a sync/async environment, see the optional dependency of `requests` below.
+
+```py
+# async example - it will default to async
+import mystbin
+
+mystbin_client = mystbin.MystClient() ## api_key kwarg for authentication also
+
+await mystbin_client.post("Hello from Mystb.in!", suffix="python")
+>>> 'https://mystb.in/<your generated ID>.python'
+```
+
+```py
+# sync example - we need to pass a session though
+import mystbin
+import requests
+
+sync_session = requests.Session()
+mystbin_client = mystbin.MystClient(session=sync_session) ## optional api_key kwarg also
+
+mystbin_client.post("Hello from sync Mystb.in!", suffix="text")
+>>> 'https://mystb.in/<your generated ID>.text'
+```
+
+NOTE: the session - aiohttp or requests - will have their default headers changed during init to support the Authorization header with the api key if present, and there is a timeout of 15s for each operation.
+
 ### Dependencies
 
-`aiohttp` - required
+`aiohttp` - required \
 `requests` - optional

mystbin/client.py

@@ -22,35 +22,140 @@
 DEALINGS IN THE SOFTWARE.
 """
 
-from typing import Optional, Union
+import asyncio
+import json
+from typing import Awaitable, Callable, Optional, Union
 
 import aiohttp
+import requests
 import yarl
 
 from .constants import *
 from .errors import *
+from .objects import *
 
 __all__ = ("MystClient", )
 
-class MystClient():
+
+class MystClient:
     """
     Client for interacting with the Mystb.in API.
 
     Attributes
     ----------
-    authorization: Optional[:class:`dict`]
-                Your username | password combination to access the Mystb.in API.
-                Can be obtained via: #TODO
     api_key: Optional[:class:`str`]
-                Your private API token to access the Mystb.in API.
-                Can be obtained via: #TODO
-    session: Optional[Union[:class:`ClientSession`, :class:`Session`]]
+        Your private API token to access the Mystb.in API.
+        Can be obtained via: #TODO
+    session: Optional[Union[:class:`aiohttp.ClientSession`, :class:`requests.Session`]]
+        Optional session to be passed to the creation of the client.
     """
-    __slots__ = ("authorization", "api_key", "session")
+    __slots__ = ("api_key", "session", "_are_we_async")
 
     def __init__(
-        self,
-        authorization: dict = None,
-        api_key: str = None, *,
-        session: Optional[Union[aiohttp.ClientSession, requests.Session]] = None,
-    )
+        self, *,
+        api_key: str = None,
+        session: Optional[Union[aiohttp.ClientSession,
+                                requests.Session]] = None
+    ) -> None:
+        self.api_key = api_key
+        self._are_we_async = session is None or isinstance(
+            session, aiohttp.ClientSession)
+        self.session = self._generate_sync_session(
+            session) if not self._are_we_async else None
+
+    def _generate_sync_session(self, session: requests.Session) -> requests.Session:
+        """ We will update a :class:`requests.Session` instance with the auth we require. """
+        # the passed session was found to be 'sync'.
+        if self.api_key:
+            session.headers.update({"Authorization": self.api_key})
+        return session
+
+    async def _generate_async_session(self, session: Optional[aiohttp.ClientSession] = None) -> aiohttp.ClientSession:
+        """ We will update (or create) a :class:`aiohttp.ClientSession` instance with the auth we require. """
+        if not session:
+            session = aiohttp.ClientSession(raise_for_status=False,
+                                            timeout=aiohttp.ClientTimeout(CLIENT_TIMEOUT))
+        if self.api_key:
+            session._default_headers.update(
+                {"Authorization": self.api_key})
+        return session
+
+    def post(self, content: str, syntax: str = None) -> Union[Paste, Awaitable]:
+        """
+        This will post to the Mystb.in API and return the url.
+        Can pass an optional suffix for the syntax highlighting.
+
+        Attributes
+        ----------
+        content: :class:`str`
+            The content you are posting to the Mystb.in API.
+        syntax: :class:`str`
+            The optional suffix to append the returned URL which is used for syntax highlighting on the paste.
+        """
+        if self._are_we_async:
+            return self._perform_async_post(content, syntax)
+        return self._perform_sync_post(content, syntax)
+
+    def _perform_sync_post(self, content: str, syntax: str = None) -> Paste:
+        """ Sync post request. """
+        payload = {'meta': [{'index': 0, 'syntax': syntax}]}
+        response: requests.Response = self.session.post(API_BASE_URL, files={
+                                                        'data': content, 'meta': (None, json.dumps(payload), 'application/json')})
+        if response.status_code not in [200, 201]:
+            raise APIError(response.status_code, response.text)
+
+        return Paste(response.json(), syntax)
+
+    async def _perform_async_post(self, content: str, syntax: str = None) -> Paste:
+        """ Async post request. """
+        if not self.session and self._are_we_async:
+            self.session = await self._generate_async_session()
+        multi_part_write = aiohttp.MultipartWriter()
+        paste_content = multi_part_write.append(content)
+        paste_content.set_content_disposition("form-data", name="data")
+        paste_content = multi_part_write.append_json(
+            {'meta': [{'index': 0, 'syntax': syntax}]}
+        )
+        paste_content.set_content_disposition("form-data", name="meta")
+        async with self.session.post(API_BASE_URL, data=multi_part_write) as response:
+            status_code = response.status
+            response_text = await response.text()
+            if status_code not in [200, 201]:
+                raise APIError(status_code, response_text)
+            response_data = await response.json()
+
+        return Paste(response_data, syntax)
+
+    def get(self, paste_id: str) -> str:
+        """
+        This will perform a GET request against the Mystb.in API and return the url.
+        Must be passed a valid paste ID or URL.
+
+        Attributes
+        ----------
+        paste_id: :class:`str`
+            The ID of the paste you are going to retrieve.
+        """
+        paste_id_match = MB_URL_RE.match(paste_id)
+        if not paste_id_match:
+            raise BadPasteID("This is an invalid Mystb.in paste ID.")
+        paste_id = paste_id_match.group('ID')
+        syntax = paste_id_match.group('syntax')
+        if not self._are_we_async:
+            return self._perform_sync_get(paste_id, syntax)
+        return self._perform_async_get(paste_id, syntax)
+
+    def _perform_sync_get(self, paste_id: str, syntax: str = None) -> PasteData:
+        """ Sync get request. """
+        response = self.session.get(
+            f"{API_BASE_URL}/{paste_id}", timeout=CLIENT_TIMEOUT)
+        paste_data = response.json()
+        return PasteData(paste_id, paste_data)
+
+    async def _perform_async_get(self, paste_id: str, syntax: str = None) -> PasteData:
+        """ Async get request. """
+        if not self.session:
+            self.session = await self._generate_async_session()
+        async with self.session.get(f"{API_BASE_URL}/{paste_id}", timeout=aiohttp.ClientTimeout(CLIENT_TIMEOUT)) as response:
+            paste_data = await response.json()
+        return PasteData(paste_id, paste_data)

mystbin/constants.py

@@ -23,12 +23,12 @@
 """
 
 from re import compile
-from aiohttp import ClientTimeout
-
 
 API_BASE_URL = "https://mystb.in/api/pastes"
+PASTE_BASE = "https://mystb.in/{}{}"
 
-CLIENT_TIMEOUT = ClientTimeout(total=15.0)
+CLIENT_TIMEOUT = 15
 
-# grab the paste id: https://regex101.com/r/qkluDh/1
-MB_URL_RE = compile(r"(?:http[s]?://mystb\.in/)?([a-zA-Z]+)(?:.*)")
+# grab the paste id: https://regex101.com/r/qkluDh/6
+MB_URL_RE = compile(
+    r"(?:(?:https?://)?mystb\.in/)?(?P<ID>[a-zA-Z]+)(?:\.(?P<syntax>[a-zA-Z0-9]+))?")

mystbin/errors.py

@@ -22,3 +22,21 @@
 DEALINGS IN THE SOFTWARE.
 """
 
+
+class BadPasteID(ValueError):
+    """ Bad Paste ID. """
+
+class MystbinException(Exception):
+    """ Error when interacting with Mystbin. """
+
+class APIError(MystbinException):
+    __slots__ = ("status_code", "message")
+    def __init__(self, status_code: int, message: str) -> None:
+        self.status_code = status_code
+        self.message = message
+
+    def __repr__(self) -> str:
+        return f"<MystbinError status_code={self.status_code} message={self.message}>"
+
+    def __str__(self) -> str:
+        return self.message

mystbin/objects.py

@@ -0,0 +1,82 @@
+"""
+The MIT License (MIT)
+
+Copyright (c) 2020 AbstractUmbra
+
+Permission is hereby granted, free of charge, to any person obtaining a
+copy of this software and associated documentation files (the "Software"),
+to deal in the Software without restriction, including without limitation
+the rights to use, copy, modify, merge, publish, distribute, sublicense,
+and/or sell copies of the Software, and to permit persons to whom the
+Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.
+"""
+
+import datetime
+from textwrap import dedent
+
+from .constants import PASTE_BASE
+
+class Paste:
+    __slots__ = ("paste_id", "nick", "syntax")
+    def __init__(self, json_data: dict, syntax: str = None) -> None:
+        self.paste_id = json_data['pastes'][0]['id']
+        self.nick = json_data['pastes'][0]['nick']
+        self.syntax = syntax
+
+    def __str__(self) -> str:
+        """ Cast the Paste to a string for the URL. """
+        return self.url
+
+    def __repr__(self) -> str:
+        """ Paste repr. """
+        return f"<Paste id={self.paste_id} nick={self.nick} syntax={self.syntax}>"
+
+    @property
+    def url(self) -> str:
+        syntax = f".{self.syntax}" if self.syntax else ""
+        return PASTE_BASE.format(self.paste_id, syntax)
+
+class PasteData:
+    __slots__ = ("paste_id", "_paste_data", "paste_content", "paste_syntax", "paste_nick", "paste_date")
+    def __init__(self, paste_id: str, paste_data: dict) -> None:
+        self.paste_id = paste_id
+        self._paste_data = paste_data
+        self.paste_content = paste_data['data']
+        self.paste_syntax = paste_data['syntax']
+        self.paste_nick = paste_data['nick']
+        self.paste_date = paste_data['created_at']
+
+    def __str__(self) -> str:
+        """ We'll return the paste content. Since it's dev stuff, dedent it. """
+        return self.content
+
+    def __repr__(self) -> str:
+        """ Paste content repr. """
+        return f"<PasteData id={self.paste_id} nick={self.paste_nick} syntax={self.paste_syntax}>"
+
+    @property
+    def url(self) -> str:
+        """ The Paste ID's URL """
+        syntax = f".{self.paste_syntax}" if self.paste_syntax else ""
+        return PASTE_BASE.format(self.paste_id, syntax)
+
+    @property
+    def created_at(self) -> datetime.datetime:
+        """ Returns a UTC datetime of when the paste was created. """
+        return datetime.datetime.strptime(self.paste_date, "%Y-%m-%dT%H:%M:%S.%f")
+
+    @property
+    def content(self) -> str:
+        """ Return the paste content but dedented correctly. """
+        return dedent(self.paste_content)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "mystbin.py"
-version = "0.1.0"
+version = "0.2.0"
 description = "A small simple wrapper around the mystb.in API."
 authors = ["AbstractUmbra <Umbra@AbstractUmbra.xyz>"]
 license = "MIT"