Skip to content

Endpoint

toggl_api.modules.meta.enums.RequestMethod

Bases: Enum

Self explanatory enumerations describing the different request types primarily for selecting request methods.

toggl_api.modules.meta.base_endpoint.TogglEndpoint

Bases: ABC

Base class with basic functionality for all API requests.

Source code in toggl_api/modules/meta/base_endpoint.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class TogglEndpoint(ABC):
    """Base class with basic functionality for all API requests."""

    OK_RESPONSE: Final[int] = 200
    NOT_FOUND: Final[int] = 404
    SERVER_ERROR: Final[int] = 5

    BASE_ENDPOINT: Final[str] = "https://api.track.toggl.com/api/v9/"
    HEADERS: Final[dict] = {"content-type": "application/json"}

    __slots__ = ("__client", "workspace_id")

    def __init__(
        self,
        workspace_id: int,
        auth: httpx.BasicAuth,
        *,
        timeout: int = 20,
        **kwargs,
    ) -> None:
        self.workspace_id = workspace_id
        # NOTE: USES BASE_ENDPOINT instead of endpoint property for base_url
        # as current httpx concatenation is causing appended slashes.
        self.__client = httpx.Client(
            base_url=self.BASE_ENDPOINT,
            timeout=timeout,
            auth=auth,
        )
        atexit.register(self.__client.close)

    def method(self, method: RequestMethod) -> Callable:
        match_dict: dict[RequestMethod, Callable] = {
            RequestMethod.GET: self.__client.get,
            RequestMethod.POST: self.__client.post,
            RequestMethod.PUT: self.__client.put,
            RequestMethod.DELETE: self.__client.delete,
            RequestMethod.PATCH: self.__client.patch,
        }
        return match_dict.get(method, self.__client.get)

    def request(
        self,
        parameters: str,
        headers: Optional[dict] = None,
        body: Optional[dict] = None,
        method: RequestMethod = RequestMethod.GET,
    ) -> Optional[list[TogglClass] | TogglClass]:
        """Main request method which handles putting together the final API
        request.

        Args:
            parameters (str): Request parameters with the endpoint excluded.
                Will concate with the endpoint property.
            headers (dict, optional): Custom request headers. Defaults to
                class property if set to None.
            body (dict, optional): Request body JSON data for specifying info.
                Defaults to None. Only used with none-GET or DELETE requests.
            method (RequestMethod): Request method to select. Defaults to GET.

        Returns:
            dict | None: Response data or None if request does not return any
                data.
        """

        url = self.endpoint + parameters
        headers = headers or self.HEADERS

        if body and method not in {RequestMethod.DELETE, RequestMethod.GET}:
            response = self.method(method)(url, headers=headers, json=body)
        else:
            response = self.method(method)(url, headers=headers)

        if response.status_code != self.OK_RESPONSE:
            # TODO: Toggl API return code lookup.
            msg = "Request failed with status code %s: %s"
            log.error(msg, response.status_code, response.text)
            if response.status_code % 100 == self.SERVER_ERROR:
                delay = random.randint(1, 5)
                log.error("Status code is a server error. Retrying request in %s seconds", delay)
                time.sleep(delay)
                return self.request(parameters, headers, body, method)

            response.raise_for_status()

        try:
            data = response.json()
        except ValueError:
            return None

        if isinstance(data, list):
            data = self.process_models(data)
        elif isinstance(data, dict):
            data = self.model.from_kwargs(**data)

        return data

    def process_models(
        self,
        data: list[dict[str, Any]],
    ) -> list[TogglClass]:
        return [self.model.from_kwargs(**mdl) for mdl in data]

    @property
    @abstractmethod
    def endpoint(self) -> str:
        pass

    @property
    @abstractmethod
    def model(self) -> type[TogglClass]:
        return TogglClass

endpoint: str abstractmethod property

model: type[TogglClass] abstractmethod property

request(parameters: str, headers: Optional[dict] = None, body: Optional[dict] = None, method: RequestMethod = RequestMethod.GET) -> Optional[list[TogglClass] | TogglClass]

Main request method which handles putting together the final API request.

Parameters:

  • parameters (str) –

    Request parameters with the endpoint excluded. Will concate with the endpoint property.

  • headers (dict, default: None ) –

    Custom request headers. Defaults to class property if set to None.

  • body (dict, default: None ) –

    Request body JSON data for specifying info. Defaults to None. Only used with none-GET or DELETE requests.

  • method (RequestMethod, default: GET ) –

    Request method to select. Defaults to GET.

Returns:

  • Optional[list[TogglClass] | TogglClass] –

    dict | None: Response data or None if request does not return any data.

Source code in toggl_api/modules/meta/base_endpoint.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def request(
    self,
    parameters: str,
    headers: Optional[dict] = None,
    body: Optional[dict] = None,
    method: RequestMethod = RequestMethod.GET,
) -> Optional[list[TogglClass] | TogglClass]:
    """Main request method which handles putting together the final API
    request.

    Args:
        parameters (str): Request parameters with the endpoint excluded.
            Will concate with the endpoint property.
        headers (dict, optional): Custom request headers. Defaults to
            class property if set to None.
        body (dict, optional): Request body JSON data for specifying info.
            Defaults to None. Only used with none-GET or DELETE requests.
        method (RequestMethod): Request method to select. Defaults to GET.

    Returns:
        dict | None: Response data or None if request does not return any
            data.
    """

    url = self.endpoint + parameters
    headers = headers or self.HEADERS

    if body and method not in {RequestMethod.DELETE, RequestMethod.GET}:
        response = self.method(method)(url, headers=headers, json=body)
    else:
        response = self.method(method)(url, headers=headers)

    if response.status_code != self.OK_RESPONSE:
        # TODO: Toggl API return code lookup.
        msg = "Request failed with status code %s: %s"
        log.error(msg, response.status_code, response.text)
        if response.status_code % 100 == self.SERVER_ERROR:
            delay = random.randint(1, 5)
            log.error("Status code is a server error. Retrying request in %s seconds", delay)
            time.sleep(delay)
            return self.request(parameters, headers, body, method)

        response.raise_for_status()

    try:
        data = response.json()
    except ValueError:
        return None

    if isinstance(data, list):
        data = self.process_models(data)
    elif isinstance(data, dict):
        data = self.model.from_kwargs(**data)

    return data

method(method: RequestMethod) -> Callable

Source code in toggl_api/modules/meta/base_endpoint.py 
58
59
60
61
62
63
64
65
66
def method(self, method: RequestMethod) -> Callable:
    match_dict: dict[RequestMethod, Callable] = {
        RequestMethod.GET: self.__client.get,
        RequestMethod.POST: self.__client.post,
        RequestMethod.PUT: self.__client.put,
        RequestMethod.DELETE: self.__client.delete,
        RequestMethod.PATCH: self.__client.patch,
    }
    return match_dict.get(method, self.__client.get)

process_models(data: list[dict[str, Any]]) -> list[TogglClass]

Source code in toggl_api/modules/meta/base_endpoint.py 
124
125
126
127
128
def process_models(
    self,
    data: list[dict[str, Any]],
) -> list[TogglClass]:
    return [self.model.from_kwargs(**mdl) for mdl in data]

toggl_api.modules.meta.cached_endpoint.TogglCachedEndpoint

Bases: TogglEndpoint

Abstract cached endpoint for requesting toggl API data to disk.

Attributes:

  • _cache –

    Cache object for caching toggl API data to disk. Builtin cache types are JSONCache and SqliteCache.

Source code in toggl_api/modules/meta/cached_endpoint.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class TogglCachedEndpoint(TogglEndpoint):
    """Abstract cached endpoint for requesting toggl API data to disk.

    Attributes:
        _cache: Cache object for caching toggl API data to disk. Builtin cache
            types are JSONCache and SqliteCache.
    """

    __slots__ = ("_cache",)

    def __init__(
        self,
        workspace_id: int,
        auth: httpx.BasicAuth,
        cache: TogglCache,
        *,
        timeout: int = 20,
        **kwargs,
    ) -> None:
        super().__init__(
            workspace_id=workspace_id,
            auth=auth,
            timeout=timeout,
            **kwargs,
        )
        self.cache = cache

    def request(  # type: ignore[override]
        self,
        parameters: str,
        headers: Optional[dict] = None,
        body: Optional[dict] = None,
        method: RequestMethod = RequestMethod.GET,
        *,
        refresh: bool = False,
    ) -> Optional[TogglClass | Iterable[TogglClass]]:
        """Overridden request method with builtin cache.

        Args:
            parameters: Request parameters with the endpoint excluded.
            headers: Request headers. Custom headers can be added here.
            body: Request body for GET, POST, PUT, PATCH requests.
                Defaults to None.
            method: Request method. Defaults to GET.
            refresh: Whether to refresh the cache or not. Defaults to False.

        Returns:
            TogglClass | Iterable[TogglClass] | None: Toggl API response data
                processed into TogglClass objects.
        """

        data = self.load_cache()
        if data and not refresh:
            log.info(
                "Loading request %s%s data from cache.",
                self.endpoint,
                parameters,
                extra={"body": body, "headers": headers, "method": method},
            )
            return data

        response = super().request(
            parameters,
            method=method,
            headers=headers,
            body=body,
        )

        if response is None or method == RequestMethod.DELETE:
            return None

        self.save_cache(response, method)  # type: ignore[arg-type]

        return response

    def load_cache(self) -> Iterable[TogglClass]:
        return self.cache.load_cache()

    def save_cache(
        self,
        response: list[TogglClass] | TogglClass,
        method: RequestMethod,
    ) -> None:
        if isinstance(self.cache.expire_after, timedelta) and not self.cache.expire_after.total_seconds():
            return None
        return self.cache.save_cache(response, method)

    def query(
        self,
        *,
        inverse: bool = False,
        distinct: bool = False,
        **query: dict[str, Any],
    ) -> Iterable[TogglClass]:
        return self.cache.query(inverse=inverse, distinct=distinct, **query)

    @property
    @abstractmethod
    def endpoint(self) -> str:
        pass

    @property
    def cache(self) -> TogglCache:
        return self._cache

    @cache.setter
    def cache(self, value: TogglCache) -> None:
        self._cache = value
        if self.cache.parent is None:
            self.cache.parent = self

cache: TogglCache property writable

request(parameters: str, headers: Optional[dict] = None, body: Optional[dict] = None, method: RequestMethod = RequestMethod.GET, *, refresh: bool = False) -> Optional[TogglClass | Iterable[TogglClass]]

Overridden request method with builtin cache.

Parameters:

  • parameters (str) –

    Request parameters with the endpoint excluded.

  • headers (Optional[dict], default: None ) –

    Request headers. Custom headers can be added here.

  • body (Optional[dict], default: None ) –

    Request body for GET, POST, PUT, PATCH requests. Defaults to None.

  • method (RequestMethod, default: GET ) –

    Request method. Defaults to GET.

  • refresh (bool, default: False ) –

    Whether to refresh the cache or not. Defaults to False.

Returns:

  • Optional[TogglClass | Iterable[TogglClass]] –

    TogglClass | Iterable[TogglClass] | None: Toggl API response data processed into TogglClass objects.

Source code in toggl_api/modules/meta/cached_endpoint.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def request(  # type: ignore[override]
    self,
    parameters: str,
    headers: Optional[dict] = None,
    body: Optional[dict] = None,
    method: RequestMethod = RequestMethod.GET,
    *,
    refresh: bool = False,
) -> Optional[TogglClass | Iterable[TogglClass]]:
    """Overridden request method with builtin cache.

    Args:
        parameters: Request parameters with the endpoint excluded.
        headers: Request headers. Custom headers can be added here.
        body: Request body for GET, POST, PUT, PATCH requests.
            Defaults to None.
        method: Request method. Defaults to GET.
        refresh: Whether to refresh the cache or not. Defaults to False.

    Returns:
        TogglClass | Iterable[TogglClass] | None: Toggl API response data
            processed into TogglClass objects.
    """

    data = self.load_cache()
    if data and not refresh:
        log.info(
            "Loading request %s%s data from cache.",
            self.endpoint,
            parameters,
            extra={"body": body, "headers": headers, "method": method},
        )
        return data

    response = super().request(
        parameters,
        method=method,
        headers=headers,
        body=body,
    )

    if response is None or method == RequestMethod.DELETE:
        return None

    self.save_cache(response, method)  # type: ignore[arg-type]

    return response

load_cache() -> Iterable[TogglClass]

Source code in toggl_api/modules/meta/cached_endpoint.py 
110
111
def load_cache(self) -> Iterable[TogglClass]:
    return self.cache.load_cache()

save_cache(response: list[TogglClass] | TogglClass, method: RequestMethod) -> None

Source code in toggl_api/modules/meta/cached_endpoint.py 
113
114
115
116
117
118
119
120
def save_cache(
    self,
    response: list[TogglClass] | TogglClass,
    method: RequestMethod,
) -> None:
    if isinstance(self.cache.expire_after, timedelta) and not self.cache.expire_after.total_seconds():
        return None
    return self.cache.save_cache(response, method)