task-team/odoo_modules/task_team_connector/services/api_client.py

# Task Team Connector — REST API client (stdlib-only, no extra deps)
import json
import logging
import urllib.error
import urllib.request

_logger = logging.getLogger(__name__)


class TaskTeamApiClient:
    """Thin HTTP wrapper for the Task Team REST API."""

    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key          # service-account e-mail
        self._token: str | None = None  # cached JWT

    # ------------------------------------------------------------------
    # Auth
    # ------------------------------------------------------------------

    def _authenticate(self):
        """Login with the service account and cache the JWT."""
        payload = json.dumps({'email': self.api_key}).encode()
        req = urllib.request.Request(
            f'{self.base_url}/auth/login',
            data=payload,
            headers={'Content-Type': 'application/json'},
            method='POST',
        )
        try:
            with urllib.request.urlopen(req, timeout=10) as resp:
                data = json.loads(resp.read())
                self._token = data.get('data', {}).get('token')
        except urllib.error.HTTPError as exc:
            _logger.error('Task Team authentication failed: %s', exc)
            self._token = None

    def _headers(self) -> dict:
        if not self._token:
            self._authenticate()
        hdrs = {'Content-Type': 'application/json'}
        if self._token:
            hdrs['Authorization'] = f'Bearer {self._token}'
        return hdrs

    # ------------------------------------------------------------------
    # Generic request
    # ------------------------------------------------------------------

    def _request(self, method: str, path: str, payload=None, *, retry=True):
        url = f'{self.base_url}{path}'
        data = json.dumps(payload).encode() if payload is not None else None
        req = urllib.request.Request(url, data=data, headers=self._headers(), method=method)
        try:
            with urllib.request.urlopen(req, timeout=15) as resp:
                return json.loads(resp.read())
        except urllib.error.HTTPError as exc:
            if exc.code == 401 and retry:
                # Token expired — re-auth once
                self._token = None
                return self._request(method, path, payload, retry=False)
            body = exc.read().decode(errors='replace') if exc else ''
            _logger.error('Task Team API %s %s → %s: %s', method, path, exc.code, body[:200])
            raise
        except Exception as exc:
            _logger.error('Task Team API request error (%s %s): %s', method, path, exc)
            raise

    # ------------------------------------------------------------------
    # User helpers
    # ------------------------------------------------------------------

    def get_or_create_user(self, email: str, name: str) -> dict:
        """Find a Task Team user by e-mail, creating them if absent."""
        payload = json.dumps({'email': email}).encode()
        req = urllib.request.Request(
            f'{self.base_url}/auth/login',
            data=payload,
            headers={'Content-Type': 'application/json'},
            method='POST',
        )
        try:
            with urllib.request.urlopen(req, timeout=10) as resp:
                return json.loads(resp.read())
        except urllib.error.HTTPError as exc:
            if exc.code == 401:
                # Not registered yet
                reg_payload = json.dumps({'email': email, 'name': name}).encode()
                reg_req = urllib.request.Request(
                    f'{self.base_url}/auth/register',
                    data=reg_payload,
                    headers={'Content-Type': 'application/json'},
                    method='POST',
                )
                with urllib.request.urlopen(reg_req, timeout=10) as resp:
                    return json.loads(resp.read())
            raise

    # ------------------------------------------------------------------
    # Task CRUD
    # ------------------------------------------------------------------

    def create_task(self, payload: dict, user_email: str | None = None) -> dict:
        return self._request('POST', '/tasks', payload)

    def update_task(self, tt_task_id: str, payload: dict, user_email: str | None = None) -> dict:
        return self._request('PUT', f'/tasks/{tt_task_id}', payload)

    def get_task(self, tt_task_id: str) -> dict:
        return self._request('GET', f'/tasks/{tt_task_id}')

    def list_tasks(self, **filters) -> dict:
        qs = '&'.join(f'{k}={v}' for k, v in filters.items() if v is not None)
        path = f'/tasks?{qs}' if qs else '/tasks'
        return self._request('GET', path)

    # ------------------------------------------------------------------
    # Webhook
    # ------------------------------------------------------------------

    def send_webhook(self, event: str, payload: dict) -> dict:
        return self._request('POST', '/connectors/webhook/odoo', {'event': event, **payload})