Initial commit

.editorconfig

@@ -0,0 +1,16 @@
+# EditorConfig is awesome: https://EditorConfig.org
+
+# top-most EditorConfig file
+root = true
+
+[*]
+indent_style = space
+indent_size = 4
+end_of_line = lf
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true
+max_line_length = 132
+
+[*.yml]
+indent_size = 2

.github/workflows/build.yml

@@ -0,0 +1,37 @@
+name: Build
+on:
+  push:
+    branches: [main]
+  workflow_dispatch:
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: 3.11
+      - name: Install poetry
+        run: pip install -q poetry==1.5.0
+      - name: Setup local virtual environment
+        run: |
+          poetry config virtualenvs.create true --local
+          poetry config virtualenvs.in-project true --local
+      - uses: actions/cache@v4
+        with:
+          path: ./.venv
+          key: venv-${{ hashFiles('poetry.lock') }}
+      - name: Install the project dependencies
+        run: poetry install -q
+      - name: Run the automated tests
+        env:
+          ADJUST_EMAIL: ${{ secrets.ADJUST_EMAIL }}
+          ADJUST_PASSWORD: ${{ secrets.ADJUST_PASSWORD }}
+        run: |
+          echo email=$ADJUST_EMAIL
+          echo password=$ADJUST_PASSWORD
+          poetry run pytest --cov=adjust --cov-report xml:cov.xml -vv
+      - name: Upload coverage reports to Codecov
+        uses: codecov/codecov-action@v4
+        with:
+          token: ${{ secrets.CODECOV_TOKEN }}

.gitignore

@@ -0,0 +1,8 @@
+.DS_Store
+__pycache__/
+.mypy_cache/
+.pytest_cache/
+*.pyc
+.env
+.coverage*
+cov.xml

.mypy.ini

@@ -0,0 +1,9 @@
+[mypy]
+files=**/*.py
+disallow_untyped_defs = True
+warn_return_any = True
+warn_unused_configs = True
+
+[mypy-tests.*]
+disallow_untyped_defs = True
+no_implicit_optional = True

.vscode/launch.json

@@ -0,0 +1,27 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "CLI: add placeholders",
+            "type": "debugpy",
+            "request": "launch",
+            "module": "adjust.cli",
+            "args": [
+                "callbacks",
+                "snapshot",
+                "placeholders",
+                "add",
+                "-u",
+                "invenio.sgn.com/v1/adjust",
+                "-a",
+                "!Cookie,!Bingo",
+                "-n",
+                "jamcity_ext",
+            ],
+            "console": "integratedTerminal",
+        },
+    ]
+}

.vscode/settings.json

@@ -0,0 +1,11 @@
+{
+    "python.testing.pytestArgs": [
+        "tests",
+        "--cov=adjust",
+        "--cov-report",
+        "xml:cov.xml",
+        "-vv"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

README.md

@@ -0,0 +1,42 @@
+# Adjust CLI
+
+## Description
+
+Adjust CLI is a Python package that provides a command-line interface (CLI) to manage Adjust callbacks. It allows you to easily configure and handle callbacks for your Adjust integration.
+
+## Features
+
+- Backup and restore callbacks configuration through snapshots
+- Add and remove placeholders to/from a subset of callbacks in a snapshot
+
+## Installation
+
+To install Adjust Callback Manager, you can use pip:
+
+```
+pip install "adjust-cli@git+https://github.com/mindjolt/adjust-cli"
+```
+
+## Usage
+
+This CLI provides several commands for managing Adjust callbacks and snapshots.
+
+### Snapshot
+
+Create a local snapshot of all Adjust callbacks:
+
+```bash
+adjust snapshot create --snapshot SNAPSHOT_PATH
+```
+
+Restore Adjust callbacks from a local snapshot:
+
+```bash
+adjust snapshot restore --snapshot SNAPSHOT_PATH
+```
+
+Add placeholders to a snapshot:
+
+```bash
+adjust snapshot modify --snapshot SNAPSHOT_PATH --having-app MyApp -a PLACEHOLDER
+```

adjust/api/__init__.py

@@ -0,0 +1,3 @@
+__all__ = ["AdjustAPI"]
+
+from .api import AdjustAPI

adjust/api/api.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python
+from __future__ import annotations
+from functools import cached_property
+import json
+import os
+from types import NoneType
+from bs4 import BeautifulSoup
+from pydantic import parse_obj_as
+from typing import Any, Optional, Type, TypeVar
+
+import requests
+
+from .model import RawCallback, App, AppsResponse, Callback, Event, EventsResponse, Placeholder, User
+
+
+T = TypeVar("T")
+
+
+class AdjustAPI(object):
+    """A class to interact with the Adjust API
+
+    Attributes:
+        user : User
+            The currently logged in user
+        placeholders : list[Placeholder]
+            A list of all the available Adjust placeholders
+        apps : list[App]
+            The list of all the registered apps on the Adjust dashboard
+            The app object contains attributes such as the adjust token or the platform
+            appIds
+
+    Methods:
+        events(app: App | str) : list[Event]
+            Get the list of all the events registered for an App or app token
+        callbacks(app: App | str) : list[Callback]
+            Get the list of all the available callbacks for an App or app token.
+            The returned Callback objects contain the endpoint URL if one is set.
+        set_callback(app: App | str, type: str, callback_url: str)
+            Updates the callback of a given type to the new callback_url.
+
+    """
+
+    def __init__(self, email: str = None, password: str = None):
+        """Creates an object that can be used to issue calls to the Adjust API.
+        Valid credentials must be supplied in order for the API calls to be
+        authenticated and authorized.
+
+        Args:
+            email (str): the authentication email
+            password (str): the authentication password
+        """
+        email = email or os.getenv("ADJUST_EMAIL")
+        if not email:
+            raise ValueError("Email not provided")
+        password = password or os.getenv("ADJUST_PASSWORD")
+        if not password:
+            raise ValueError("Password not provided")
+        self._session = requests.Session()
+        self._user: Optional[User] = None
+        self._log_in = lambda: self._sign_in(email, password)
+        self._logged_in = False
+
+    def _log_in_if_needed(self) -> None:
+        """Internal method used to log in upon first use"""
+        if not self._logged_in:
+            self._logged_in = True
+            self._log_in()
+
+    def _api(self, type: Type[T], path: str, method: str = "GET", **data: Any) -> T:
+        """Internal method used to emit low-level API calls.
+
+        Args:
+            path (str): the API path to call
+            method (str, optional): The HTTP method to use. Defaults to "GET".
+
+        Returns:
+            Any: the API response
+        """
+        self._log_in_if_needed()
+        url = "https://api.adjust.com/" + path
+        headers = dict(Accept="application/json")
+        if not data:
+            r = self._session.get(url, headers=headers)
+        elif method == "PUT":
+            r = self._session.put(url, headers=headers, json=data)
+        else:
+            r = self._session.post(url, headers=headers, json=data)
+        r.raise_for_status()
+        return parse_obj_as(type, None if r.status_code == 204 else r.json())
+
+    def _sign_in(self, email: str, password: str) -> None:
+        """Internal method to authenticate with the Adjust API
+
+        Args:
+            email (str): The login email
+            password (str): The login password
+
+        Returns:
+            User: The logged in user
+        """
+        user = dict(email=email, password=password, remember_me=True)
+        self._user = self._api(User, "accounts/users/sign_in", user=user)
+
+    def user(self) -> Optional[User]:
+        """Returns the currently logged in user
+
+        Returns:
+            User
+        """
+        self._log_in_if_needed()
+        return self._user
+
+    @cached_property
+    def placeholders(self) -> list[Placeholder]:
+        """Returns a list of all available Adjust placeholders
+
+        Returns:
+            list[Placeholder]
+        """
+        url = "https://help.adjust.com/en/partner/placeholders"
+        html = self._session.get(url).content
+        soup = BeautifulSoup(html, "lxml")
+        script = soup.find(id="__NEXT_DATA__")
+        assert script, "Could not find placeholders data"
+        data = json.loads(script.text)
+        placeholders = data["props"]["pageProps"]["placeholdersData"]
+        return [Placeholder.parse_obj(p) for p in placeholders]
+
+    @cached_property
+    def apps(self) -> list[App]:
+        """Returns a list of all the registered apps on the Adjust dashboard.
+        The returned list contains one object per application, with properties
+        such as the app token or platform appIds.
+
+        Returns:
+            list[App]
+        """
+        response = self._api(AppsResponse, "dashboard/api/apps")
+        return response.apps
+
+    def callbacks(self, app: App | str) -> list[Callback]:
+        """Returns the list of callbacks available for an app or app token.
+
+        Args:
+            app (App | str): the app or app token
+
+        Returns:
+            list[Callback]: the callback mapping
+        """
+        token = app if isinstance(app, str) else app.token
+        cbs = self._api(list[RawCallback], f"dashboard/api/apps/{token}/callbacks")
+        return [c.to_callback() for c in cbs]
+
+    def events(self, app: App | str, include_archived: bool = False) -> dict[str, Event]:
+        """Returns a mapping of all the events for an app or app token.
+        Events are mapped by their event token.
+
+        Args:
+            app (App | str): The app or app token
+            include_archived (bool, optional): True to include archived events.
+                                               Defaults to False.
+
+        Returns:
+            list[Event]: the list of events
+        """
+        token = app if isinstance(app, str) else app.token
+        template = f"dashboard/api/apps/{token}/event_types?include_archived={include_archived}"  # noqa: E501
+        data = self._api(EventsResponse, template)
+        return {e.token: e for e in data.events}
+
+    def update_callback(self, app: App | str, callback: Callback) -> None:
+        """Updates an app callback.
+
+        Args:
+            app (App | str): The app or app token
+            callback (Callback): The modified callback to be updated
+        """
+        token = app if isinstance(app, str) else app.token
+        path = f"dashboard/api/apps/{token}/event_types/{callback.id}/callback"
+        self._api(NoneType, path, method="PUT", callback_url=callback.url)