Starter for project (a bit of left over from the template at this poi…

…nt).

.gitignore

@@ -157,4 +157,5 @@ cython_debug/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+.idea/
+

LICENSE

@@ -1,4 +1,4 @@
-MIT License
+The MIT License (MIT)
 
 Copyright (c) 2024 Michael Kennedy
 
@@ -18,4 +18,4 @@ 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.
+SOFTWARE.

README.md

@@ -1,2 +1,22 @@
-# listmonk
-Listmonk Email App API Client for Python
+# Listmonk Email App API Client for Python
+
+Client for the for privacy-preserving, open source [Listmonk email platform](https://listmonk.app) based on 
+`httpx` and `pydantic`. 
+
+## Partially implemented
+
+Implemented endpoints:
+
+* `POST /api/auth/login` as `login_async` and `login` ...
+
+## Installation
+
+Just `pip install listmonk`
+
+
+## Usage
+
+```python
+...
+```
+

example_client/client.py

@@ -0,0 +1,49 @@
+import json
+from pathlib import Path
+
+import umami
+
+file = Path(__file__).parent / 'settings.json'
+
+settings = {}
+if file.exists():
+    settings = json.loads(file.read_text())
+
+print(umami.user_agent)
+
+url = settings.get('base_url') or input("Enter the base URL for your instance: ")
+user = settings.get('username') or input("Enter the username for Umami: ")
+password = settings.get('password') or input("Enter the password for ")
+
+umami.set_url_base(url)
+login = umami.login(user, password)
+print(f"Logged in successfully as {login.user.username} : admin? {login.user.isAdmin}")
+print()
+
+print("Verify token:")
+print(umami.verify_token())
+print()
+
+websites = umami.websites()
+print(f"Found {len(websites):,} websites.")
+print("First website in list:")
+print(websites[0])
+print()
+
+if test_domain := settings.get('test_domain'):
+
+    test_site = [w for w in websites if w.domain == test_domain][0]
+    print(f"Using {test_domain} for testing events.")
+
+    event_resp = umami.new_event(
+        website_id=test_site.id,
+        event_name='Umami-Test-Event3',
+        title='Umami-Test-Event3',
+        hostname=test_site.domain,
+        url='/users/actions',
+        custom_data={'client': 'umami-tester-v1'},
+        referrer='https://talkpython.fm')
+
+    print(f"Created new event: {event_resp}")
+else:
+    print("No test domain, skipping event creation.")

example_client/settings-template.json

@@ -0,0 +1,8 @@
+{
+  "base_url": "Where your instance hosted? e.g. https://analytics.yoursite.com",
+  "username": "user_at_umami_instance",
+  "password": "password_at_umami_instance",
+  "test_domain": "domain name of the site you created for testing (will create events here)",
+
+  "ACTION": "COPY TO settings.json (excluded from git) and enter your info THERE (NOT HERE)"
+}

example_client/settings.json

@@ -0,0 +1,6 @@
+{
+  "base_url": "https://uma.talkpython.fm",
+  "username": "mkennedy",
+  "password": "gta-WCW9bwj8jpf1xaq",
+  "test_domain": "tesingapi.com"
+}

listmonk/__init__.py

@@ -0,0 +1,20 @@
+from umami import impl
+from . import models  # noqa: F401, E402
+from .impl import login_async, login  # noqa: F401, E402
+from .impl import new_event_async, new_event  # noqa: F401, E402
+from .impl import set_url_base, set_website_id, set_hostname  # noqa: F401, E402
+from .impl import verify_token_async, verify_token  # noqa: F401, E402
+from .impl import websites_async, websites  # noqa: F401, E402
+
+__author__ = 'Michael Kennedy <michael@talkpython.fm>'
+__version__ = impl.__version__
+user_agent = impl.user_agent
+
+__all__ = [
+    models,
+    set_url_base, set_website_id, set_hostname,
+    login_async, login,
+    websites_async, websites,
+    new_event_async, new_event,
+    verify_token_async, verify_token,
+]

listmonk/impl/__init__.py

@@ -0,0 +1,251 @@
+import sys
+from pprint import pprint
+from typing import Optional
+
+import httpx
+
+from umami import models, urls  # noqa: F401
+
+__version__ = '0.1.9'
+
+url_base: Optional[str] = None
+auth_token: Optional[str] = None
+default_website_id: Optional[str] = None
+default_hostname: Optional[str] = None
+event_user_agent = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:121.0) Gecko/20100101 Firefox/121.0'
+user_agent = (f'Umami-Client v{__version__} / '
+              f'Python {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}')
+
+
+def set_url_base(url: str):
+    if not url or not url.strip():
+        raise Exception("URL must not be empty")
+
+    global url_base
+    url_base = url.strip()
+
+
+def set_website_id(website: str):
+    global default_website_id
+    default_website_id = website
+
+
+def set_hostname(hostname: str):
+    global default_hostname
+    default_hostname = hostname
+
+
+async def login_async(username: str, password: str) -> models.LoginResponse:
+    global auth_token
+    validate_state(url=True)
+    validate_login(username, password)
+
+    url = f'{url_base}{urls.login}'
+    headers = {'User-Agent': user_agent}
+    api_data = {
+        "username": username,
+        "password": password,
+    }
+    async with httpx.AsyncClient() as client:
+        resp = await client.post(url, data=api_data, headers=headers, follow_redirects=True)
+        resp.raise_for_status()
+
+    model = models.LoginResponse(**resp.json())
+    auth_token = model.token
+    return model
+
+
+def login(username: str, password: str) -> models.LoginResponse:
+    global auth_token
+
+    validate_state(url=True)
+    validate_login(username, password)
+
+    url = f'{url_base}{urls.login}'
+    headers = {'User-Agent': user_agent}
+    api_data = {
+        "username": username,
+        "password": password,
+    }
+    resp = httpx.post(url, data=api_data, headers=headers, follow_redirects=True)
+    resp.raise_for_status()
+
+    model = models.LoginResponse(**resp.json())
+    auth_token = model.token
+    return model
+
+
+async def websites_async() -> list[models.Website]:
+    global auth_token
+    validate_state(url=True, user=True)
+
+    url = f'{url_base}{urls.websites}'
+    headers = {
+        'User-Agent': user_agent,
+        'Authorization': f'Bearer {auth_token}',
+    }
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(url, headers=headers, follow_redirects=True)
+        resp.raise_for_status()
+
+    model = models.WebsitesResponse(**resp.json())
+    return model.websites
+
+
+def websites() -> list[models.Website]:
+    global auth_token
+    validate_state(url=True, user=True)
+
+    url = f'{url_base}{urls.websites}'
+    headers = {
+        'User-Agent': user_agent,
+        'Authorization': f'Bearer {auth_token}',
+    }
+    resp = httpx.get(url, headers=headers, follow_redirects=True)
+    resp.raise_for_status()
+
+    data = resp.json()
+    model = models.WebsitesResponse(**data)
+    return model.websites
+
+
+async def new_event_async(event_name: str, hostname: Optional[str] = None, url: str = '/',
+                          website_id: Optional[str] = None, title: Optional[str] = None,
+                          custom_data=None, referrer: str = '', language: str = 'en-US',
+                          screen: str = "1920x1080") -> str:
+    website_id = website_id or default_website_id
+    hostname = hostname or default_hostname
+    title = title or event_name
+    custom_data = custom_data or {}
+
+    api_url = f'{url_base}{urls.events}'
+    headers = {
+        'User-Agent': event_user_agent,
+        'Authorization': f'Bearer {auth_token}',
+    }
+
+    payload = {
+        "hostname": hostname,
+        "language": language,
+        "referrer": referrer,
+        "screen": screen,
+        "title": title,
+        "url": url,
+        "website": website_id,
+        "name": event_name,
+        "data": custom_data
+    }
+
+    event_data = {
+        'payload': payload,
+        'type': 'event'
+    }
+
+    print("POSTING NEW EVENT")
+    print()
+    print("URL:")
+    pprint(api_url)
+    print()
+    print("Headers:")
+    pprint(headers)
+    print()
+    print("event_data:")
+    pprint(event_data)
+
+    async with httpx.AsyncClient() as client:
+        resp = await client.post(api_url, json=event_data, headers=headers, follow_redirects=True)
+        resp.raise_for_status()
+
+    return resp.text
+
+
+def new_event(event_name: str, hostname: Optional[str] = None, url: str = '/event-api-endpoint',
+              website_id: Optional[str] = None, title: Optional[str] = None,
+              custom_data=None, referrer: str = '', language: str = 'en-US',
+              screen: str = "1920x1080") -> str:
+    website_id = website_id or default_website_id
+    hostname = hostname or default_hostname
+    title = title or event_name
+    custom_data = custom_data or {}
+
+    api_url = f'{url_base}{urls.events}'
+    headers = {
+        'User-Agent': event_user_agent,
+        'Authorization': f'Bearer {auth_token}',
+    }
+
+    payload = {
+        "hostname": hostname,
+        "language": language,
+        "referrer": referrer,
+        "screen": screen,
+        "title": title,
+        "url": url,
+        "website": website_id,
+        "name": event_name,
+        "data": custom_data
+    }
+
+    event_data = {
+        'payload': payload,
+        'type': 'event'
+    }
+
+    resp = httpx.post(api_url, json=event_data, headers=headers, follow_redirects=True)
+    resp.raise_for_status()
+
+    return resp.text
+
+
+async def verify_token_async() -> bool:
+    # noinspection PyBroadException
+    try:
+        global auth_token
+        validate_state(url=True, user=True)
+
+        url = f'{url_base}{urls.verify}'
+        headers = {
+            'User-Agent': event_user_agent,
+            'Authorization': f'Bearer {auth_token}',
+        }
+        async with httpx.AsyncClient() as client:
+            resp = await client.post(url, headers=headers, follow_redirects=True)
+            resp.raise_for_status()
+
+        return 'username' in resp.json()
+    except Exception:
+        return False
+
+
+def verify_token() -> bool:
+    # noinspection PyBroadException
+    try:
+        global auth_token
+        validate_state(url=True, user=True)
+
+        url = f'{url_base}{urls.verify}'
+        headers = {
+            'User-Agent': event_user_agent,
+            'Authorization': f'Bearer {auth_token}',
+        }
+        resp = httpx.post(url, headers=headers, follow_redirects=True)
+        resp.raise_for_status()
+
+        return 'username' in resp.json()
+    except Exception:
+        return False
+
+
+def validate_login(email, password):
+    if not email:
+        raise Exception("Email cannot be empty")
+    if not password:
+        raise Exception("Password cannot be empty")
+
+
+def validate_state(url=False, user=False):
+    if url and not url_base:
+        raise Exception("URL Base must be set to proceed.")
+
+    if user and not auth_token:
+        raise Exception("You must login before proceeding.")