diff --git a/README.md b/README.md index c57c2f1..a986832 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ [![Semantic Versions](https://img.shields.io/badge/%F0%9F%9A%80-semantic%20versions-informational.svg)](https://github.com/chekos/datawrapper/releases) [![License](https://img.shields.io/github/license/chekos/datawrapper)](https://github.com/chekos/datawrapper/blob/master/LICENSE) -A light-weight python wrapper for the Datawrapper API (v3). While it is not developed by Datawrapper officially, you can use it with your API credentials from datawrapper.de +A lightweight Python wrapper for the Datawrapper API diff --git a/datawrapper/__main__.py b/datawrapper/__main__.py index 0c6006e..296d1dd 100644 --- a/datawrapper/__main__.py +++ b/datawrapper/__main__.py @@ -1,17 +1,4 @@ -"""Access Datawrapper's API to create, update, delete charts. - -Datawrapper API lets you programatically interface with your charts. -It lets you create and edit charts, update your account information and many more things - to come. - - This package is a light-weight wrapper around Datawrapper's API. - - Typical usage example: - - dw = Datawrapper(access_token = ) - - dw.account_info() -""" +"""A lightweight Python wrapper for the Datawrapper API.""" from __future__ import annotations import json @@ -21,60 +8,66 @@ from pathlib import Path from typing import Any -import IPython import pandas as pd import requests as r from IPython.display import HTML, Image -from .exceptions import FailedRequest +from .exceptions import FailedRequest, InvalidRequest logger = logging.getLogger(__name__) class Datawrapper: - """Handles connecting with Datawrapper's API. - - Handles access to your Datawrapper's account, create, delete and move charts, tables - or maps. - Will attempt to read environment variable DATAWRAPPER_ACCESS_TOKEN by default. + """Handles working with the Datawrapper API. - Args: - access_token: A personal access token to use the API. - See app.datawrapper.de/account/api-tokens. + Your main interface for configuring Datawrapper, creating, editing and + publishing charts, maps and tables. """ - _BASE_URL = "https://api.datawrapper.de" - _API_TOKEN_URL = _BASE_URL + "/v3/auth/tokens" - _ME_URL = _BASE_URL + "/v3/me" - _CHARTS_URL = _BASE_URL + "/v3/charts" - _PUBLISH_URL = _BASE_URL + "/charts" - _BASEMAPS_URL = _BASE_URL + "/v3/basemaps" - _FOLDERS_URL = _BASE_URL + "/v3/folders" - _LOGIN_URL = _BASE_URL + "/v3/auth/login" - _LOGIN_SCOPES_URL = _BASE_URL + "/v3/auth/token-scopes" - _LOGIN_TOKENS_URL = _BASE_URL + "/v3/auth/login-tokens" - _OEMBED_URL = _BASE_URL + "/v3/oembed" - _RIVER_URL = _BASE_URL + "/v3/river" - _TEAMS_URL = _BASE_URL + "/v3/teams" - _THEMES_URL = _BASE_URL + "/v3/themes" - _USERS_URL = _BASE_URL + "/v3/users" - - _ACCESS_TOKEN = os.getenv("DATAWRAPPER_ACCESS_TOKEN") + _BASE_URL = "https://api.datawrapper.de" #: The base URL for all API methods + _API_TOKEN_URL = ( + _BASE_URL + "/v3/auth/tokens" + ) #: The endpoint for API token methods + _ME_URL = ( + _BASE_URL + "/v3/me" + ) #: The endpoint for methods related to the logged in user + _CHARTS_URL = ( + _BASE_URL + "/v3/charts" + ) #: The endpoint for methods related to charts + _BASEMAPS_URL = _BASE_URL + "/v3/basemaps" #: The endpoint for basemap methods + _FOLDERS_URL = _BASE_URL + "/v3/folders" #: The endpoint for folder methods + _LOGIN_URL = _BASE_URL + "/v3/auth/login" #: The endpoint for login methods + _LOGIN_SCOPES_URL = ( + _BASE_URL + "/v3/auth/token-scopes" + ) #: The endpoint for login scopes + _LOGIN_TOKENS_URL = ( + _BASE_URL + "/v3/auth/login-tokens" + ) #: The endpoint for login tokens + _OEMBED_URL = _BASE_URL + "/v3/oembed" #: The endpoint for oembed methods + _RIVER_URL = _BASE_URL + "/v3/river" #: The endpoint for river methods + _TEAMS_URL = _BASE_URL + "/v3/teams" #: The endpoint for team methods + _THEMES_URL = _BASE_URL + "/v3/themes" #: The endpoint for theme methods + _USERS_URL = _BASE_URL + "/v3/users" #: The endpoint for user methods + + _ACCESS_TOKEN = os.getenv("DATAWRAPPER_ACCESS_TOKEN") #: The access token to use def __init__(self, access_token=_ACCESS_TOKEN): - """To create a token head to app.datawrapper.de/account/api-tokens. - - By default this will look for DATAWRAPPER_ACCESS_TOKEN environment variable. + """Initalize a connection with the Datawrapper API. Parameters ---------- - access_token : [type], optional - [description], by default _ACCESS_TOKEN + access_token : str, optional + The access token to use, by default it will look for DATAWRAPPER_ACCESS_TOKEN environment variable. + To create a token head to app.datawrapper.de/account/api-tokens. """ self._access_token = access_token self._auth_header = {"Authorization": f"Bearer {access_token}"} + # + # Web request methods + # + def delete(self, url: str, timeout: int = 15) -> bool: """Make a DELETE request to the Datawrapper API. @@ -143,9 +136,9 @@ def get(self, url: str, params: dict | None = None, timeout: int = 15) -> Any: # If it's a csv, read the text into a dataframe elif "text/csv" in response.headers["content-type"]: return pd.read_csv(StringIO(response.text)) - # Otherwise just return the text + # Otherwise just return the content else: - return response.text + return response.content # If not, raise an exception else: logger.error(f"Get request failed with status code {response.status_code}.") @@ -170,6 +163,11 @@ def patch( The timeout for the request in seconds, by default 15 extra_headers : dict, optional A dictionary of extra headers to pass to the request, by default None + + Returns + ------- + dict + A dictionary containing the response from the API. """ # Set headers headers = self._auth_header @@ -223,8 +221,9 @@ def post( Returns ------- - dict - A dictionary containing the response from the API. + dict | bool + A dictionary containing the response from the API or True if the request was + successful but did not return any data. """ # Set headers headers = self._auth_header @@ -264,6 +263,7 @@ def put( data: dict | None = None, timeout: int = 15, extra_headers: dict | None = None, + dump_data: bool = True, ) -> bool: """Make a PUT request to the Datawrapper API. @@ -277,6 +277,8 @@ def put( The timeout for the request in seconds, by default 15 extra_headers : dict, optional A dictionary of extra headers to pass to the request, by default None + dump_data: bool, optional + Whether to dump the data to json, by default True Returns ------- @@ -296,7 +298,10 @@ def put( # Convert data to json if data: - kwargs["data"] = json.dumps(data) + if dump_data: + kwargs["data"] = json.dumps(data) + else: + kwargs["data"] = data # Make the request response = r.put(url, **kwargs) @@ -308,300 +313,313 @@ def put( logger.error(f"Put request failed with status code {response.status_code}.") raise FailedRequest(response) - def account_info(self) -> dict: - """A deprecated method for calling get_my_account.""" - # Issue a deprecation warning - logger.warning( - "This method is deprecated and will be removed in a future version. " - "Use get_account_info instead." - ) + # + # Login token actions + # - # Use the newer method - return self.get_my_account() + def get_login_tokens( + self, + limit: int = 100, + offset: int = 0, + ) -> dict: + """Retrieves all login tokens associated to the current user. - def get_my_account(self) -> dict: - """Access your account information. + Parameters + ---------- + limit : int, optional + Maximum items to fetch, by default 100. Useful for pagination. + offset : int, optional + Offset for pagination, by default 0. Returns ------- dict - A dictionary containing your account information. + A dictionary containing the login tokens for your Datawrapper account. """ - return self.get(self._ME_URL) + _query: dict = {} + if limit: + _query["limit"] = limit + if offset: + _query["offset"] = offset - def update_my_account( - self, - name: str | None = None, - email: str | None = None, - role: str | None = None, - language: str | None = None, - password: str | None = None, - old_password: str | None = None, - ) -> dict: - """Update your account information. + return self.get(self._LOGIN_TOKENS_URL, params=_query) - Parameters - ---------- - name : str, optional - Your new name, by default None - email : str, optional - Your new email, by default None - role : str, optional - Your new role, by default None - language: str, optional - Your new language, by default None - password: str, optional - Your new, strong password, by default None - old_password: str, optional - Your previous password, by default None + def create_login_token(self) -> dict: + """Creates a new login token to authenticate a user, for use in CMS integrations. + + Login tokens are valid for five minutes and can only be used once. Returns ------- dict - A dictionary containing your updated account information. + A dictionary containing the login token's information. """ - _query: dict = {} - if name: - _query["name"] = name - if email: - _query["email"] = email - if role: - _query["role"] = role - if language: - _query["language"] = language - if password and old_password: - _query["password"] = password - _query["oldPassword"] = old_password - if password and not old_password: - msg = "You must provide your old password to change it." - logger.error(msg) - raise Exception(msg) - if old_password and not password: - msg = "You must provide a new password to change it." - logger.error(msg) - raise Exception(msg) - - return self.patch( - self._ME_URL, - data=_query, + response = self.post( + self._LOGIN_TOKENS_URL, + extra_headers={"content-type": "application/json"}, ) + assert isinstance(response, dict) + return response - def update_my_settings( - self, - active_team: str | None = None, - ) -> dict: - """Update your account information. + def delete_login_token(self, token_id: str | int) -> bool: + """Deletes a login token. Parameters ---------- - active_team: str, optional - Your active team + token_id : str | int + ID of login token to delete. Returns ------- - dict - The user settings dictionary following the change. + bool + True if the login token was deleted successfully. """ - _query: dict = {} - if active_team: - _query["activeTeam"] = active_team - - if not _query: - msg = "No updates submitted." - logger.error(msg) - raise Exception(msg) + return self.delete(f"{self._LOGIN_TOKENS_URL}/{token_id}") - return self.patch( - f"{self._ME_URL}/settings", - data=_query, - ) + def login(self, token: str) -> str: + """Login using a one-time login token and redirect to the URL associated with the token. - def get_my_recently_edited_charts( - self, - limit: str | int = 100, - offset: str | int = 0, - min_last_edit_step: str | int = 0, - ) -> dict: - """Get a list of your recently edited charts. + For use in CMS integrations. Parameters ---------- - limit: str | int - Maximum items to fetch. Useful for pagination. 100 by default. - offset: str | int - Number of items to skip. Useful for pagination. Zero by default. - min_last_edit_step: str | int - Filter visualizations by the last editor step they've - been opened in (1=upload, 2=describe, 3=visualize, etc). - Zero by default. + token : str + Login token. Returns ------- - dict - A dictionary with the list of charts and metadata about the selection. + str + The HTML of the page that the token redirects to. """ - _query: dict = {} - if limit: - _query["limit"] = limit - if offset: - _query["offset"] = offset - if min_last_edit_step: - _query["minLastEditStep"] = min_last_edit_step + return self.get(f"{self._LOGIN_URL}/{token}") - return self.get( - self._ME_URL + "/recently-edited-charts", - params=_query, - ) + # + # API token methods + # - def get_my_recently_published_charts( - self, - limit: str | int = 100, - offset: str | int = 0, - min_last_edit_step: str | int = 0, - ) -> dict: - """Get a list of your recently published charts. + def get_api_tokens(self, limit: int = 100, offset: int = 0) -> dict: + """Retrieves all API tokens associated to the current user. + + Response will not include full tokens for security reasons. Requires scope `auth:read`. Parameters ---------- - limit: str | int - Maximum items to fetch. Useful for pagination. 100 by default. - offset: str | int - Number of items to skip. Useful for pagination. Zero by default. - min_last_edit_step: str | int - Filter visualizations by the last editor step they've - been opened in (1=upload, 2=describe, 3=visualize, etc). - Zero by default. + limit : int, optional + Maximum items to fetch, by default 100. Useful for pagination. + offset : int, optional + Offset for pagination, by default 0. Returns ------- dict - A dictionary with the list of charts and metadata about the selection. + A dictionary containing the API tokens for your Datawrapper account. """ _query: dict = {} if limit: _query["limit"] = limit if offset: _query["offset"] = offset - if min_last_edit_step: - _query["minLastEditStep"] = min_last_edit_step - return self.get( - self._ME_URL + "/recently-published-charts", - params=_query, - ) + return self.get(self._API_TOKEN_URL, params=_query) - def get_themes( - self, limit: str | int = 100, offset: str | int = 0, deleted: bool = False - ) -> dict: - """Get a list of themes in your Datawrapper account. + def create_api_token(self, comment: str, scopes: list[str]) -> dict: + """Create a new API Token. + + Make sure to save the token somewhere, since you won't be able to see it again. Requires scope `auth:write`. Parameters ---------- - limit: str | int - Maximum items to fetch. Useful for pagination. Default 100. - offset: str | int - Number of items to skip. Useful for pagination. Default zero. - deleted: bool - Whether to include deleted themes + comment : str + Comment to describe the API token. Tip: Use something to remember where this specific token is used. + scopes : list[str] + List of scopes for the API token. Returns ------- dict - A dictionary containing the themes in your Datawrapper account. + A dictionary containing the API token's information. """ - _query = { - "limit": limit, - "offset": offset, - "deleted": json.dumps(deleted), - } - - return self.get( - self._THEMES_URL, - params=_query, + response = self.post( + self._API_TOKEN_URL, + data={"comment": comment, "scopes": scopes}, + extra_headers={"content-type": "application/json"}, ) + assert isinstance(response, dict) + return response - def add_data(self, chart_id: str, data: pd.DataFrame | str) -> r.Response: - """Add data to a specified chart. + def update_api_token( + self, id: str | int, comment: str, scopes: list[str] | None = None + ) -> bool: + """Updates an existing API token. Parameters ---------- - chart_id : str - ID of chart, table or map to add data to. - data : pd.DataFrame | str - A pandas dataframe containing the data to be added or a string that contains - the data. + id : str | int + ID of API token to update. + comment : str + Comment to describe the API token. Tip: Use something to remember where this specific token is used. + scopes : list[str], optional + List of scopes for the API token. Returns ------- - requests.Response - A requests.Response + bool + True if the API token was updated successfully. """ - # Set headers - _header = self._auth_header - _header["content-type"] = "text/csv" + _query: dict = {"comment": comment} + if scopes: + _query["scopes"] = scopes - # If data is a pandas dataframe, convert to csv - if isinstance(data, pd.DataFrame): - _data = data.to_csv(index=False, encoding="utf-8") - # If data is a string, use that - else: - _data = data + return self.put( + f"{self._API_TOKEN_URL}/{id}", + data=_query, + extra_headers={"content-type": "application/json"}, + ) - # Add data to chart - return r.put( - url=f"{self._CHARTS_URL}/{chart_id}/data", - headers=_header, - data=_data.encode("utf-8"), + def delete_api_token(self, token_id: str | int) -> bool: + """Deletes an API token. + + Parameters + ---------- + token_id : str | int + ID of API token to delete. + + Returns + ------- + bool + True if the API token was deleted successfully. + """ + return self.delete(f"{self._API_TOKEN_URL}/{token_id}") + + def get_token_scopes(self) -> list[str]: + """Get the scopes that are available to the current user. + + Returns + ------- + list[str] + A list containing the scopes available to the current user. + """ + return self.get(self._LOGIN_SCOPES_URL) + + # + # Basemap actions + # + + def get_basemaps(self) -> list[dict]: + """Get a list of the available basemaps. + + Returns + ------- + list[dict] + A list of dictionaries containing the basemaps available in your Datawrapper account. + """ + return self.get(self._BASEMAPS_URL) + + def get_basemap(self, basemap_id: str, wgs84: bool = False) -> dict: + """Get the metadata of the requested basemap. + + Parameters + ---------- + basemap_id : str + ID of basemap to get. + wgs84 : bool, optional + Whether to return the basemap in the WGS84 project, by default False + + Returns + ------- + dict + A dictionary containing the requested basemap's metadata. + """ + return self.get( + f"{self._BASEMAPS_URL}/{basemap_id}", + params={"wgs84": wgs84}, ) - def refresh_data(self, chart_id: str) -> dict: - """Fetch configured external data and add it to the chart. + def get_basemap_key(self, basemap_id: str, basemap_key: str) -> dict: + """Get the list of available values for a basemap's key. + + Parameters + ---------- + basemap_id : str + ID of basemap to get. + basemap_key : str + Metadata key of basemap to get. + + Returns + ------- + dict + A dictionary containing the requested data. + """ + return self.get(f"{self._BASEMAPS_URL}/{basemap_id}/{basemap_key}") + + # + # Charts methods + # + + def get_chart(self, chart_id: str) -> dict: + """Retrieve information of a specific chart, table or map. Parameters ---------- chart_id : str - ID of chart, table or map to add data to. + ID of chart, table, or map. Returns ------- dict - A dictionary containing the chart's information. + A dictionary containing the information of the chart, table, or map. """ - response = self.post(f"{self._CHARTS_URL}/{chart_id}/data/refresh") - assert isinstance(response, dict) - return response + return self.get(f"{self._CHARTS_URL}/{chart_id}") + + def chart_properties(self, chart_id: str) -> dict: + """A deprecated method of the get_chart method.""" + # Issue a deprecation warning + logger.warning( + "This method is deprecated and will be removed in a future version. " + "Use get_chart instead." + ) + + # Use the newer method + return self.get_chart(chart_id) def create_chart( self, - title: str = "New Chart", - chart_type: str = "d3-bars-stacked", + title: str, + chart_type: str, + theme: str | None = None, data: pd.DataFrame | str | None = None, - folder_id: str = "", - organization_id: str = "", + external_data_url: str | None = None, + folder_id: int | None = None, + organization_id: str | None = None, + forkable: bool | None = None, + language: str | None = None, metadata: dict | None = None, ) -> dict: """Creates a new Datawrapper chart, table or map. - You can pass a pandas DataFrame as a `data` argument to upload data. - - Returns the created chart's information. - Parameters ---------- - title : str, optional + title : str Title for new chart, table or map, by default "New Chart" - chart_type : str, optional - Chart type to be created. See https://developer.datawrapper.de/docs/chart-types, - by default "d3-bars-stacked" - data : [type], optional + chart_type : str + Chart type to be created. See https://developer.datawrapper.de/docs/chart-types + theme : str, optional + Theme to use for new chart, table or map, by default None + data : pd.DataFrame | str, optional A pandas DataFrame or string containing the data to be added, by default None - folder_id : str, optional + external_data_url: str, optional + URL to external data to be added to the chart, table or map, + folder_id : int, optional ID of folder in Datawrapper.de for the chart, table or map to be created in, by default "" organization_id : str, optional ID of the team where the chart should be created. The authenticated user must have access to this team. + forkable : bool, optional + Whether the chart should be forkable or not, by default None + language: str, optional + Locale of the chart (i.e. en-US), by default None metadata: dict, optional A Python dictionary of properties to add. @@ -611,55 +629,173 @@ def create_chart( A dictionary containing the created chart's information. """ # Set chart properties - _data = {"title": title, "type": chart_type} + _query: dict[str, Any] = {"title": title, "type": chart_type} + if theme: + _query["theme"] = theme if folder_id: - _data["folderId"] = folder_id + _query["folderId"] = folder_id if organization_id: - _data["organizationId"] = organization_id + _query["organizationId"] = organization_id + if forkable: + _query["forkable"] = json.dumps(forkable) + if language: + _query["language"] = language + if external_data_url: + _query["externalData"] = external_data_url if metadata: - _data["metadata"] = metadata # type: ignore + _query["metadata"] = metadata # Create chart - chart_info = self.post( + obj = self.post( self._CHARTS_URL, - data=_data, + data=_query, extra_headers={"content-type": "application/json"}, ) - assert isinstance(chart_info, dict) + assert isinstance(obj, dict) + + # Add data, if provided + if data: + self.add_data(chart_id=obj["id"], data=data) + + # Return the result + return obj + + def update_chart( + self, + chart_id: str, + title: str | None = None, + chart_type: str | None = None, + theme: str | None = None, + data: pd.DataFrame | str | None = None, + external_data_url: str | None = None, + folder_id: int | None = None, + organization_id: str | None = None, + forkable: bool | None = None, + language: str | None = None, + metadata: dict | None = None, + ) -> dict: + """Updates a chart's title, theme, type, language, folder or organization. + + Parameters + ---------- + chart_id: str + ID Of chart, table, or map. + title: str, optional + New title + chart_type: str, optional + New chart type. See https://developer.datawrapper.de/docs/chart-types + theme: str, optional + New theme + data: pd.DataFrame | str, optional + A pandas DataFrame or string containing the data to be added, + by default None + external_data_url: str, optional + URL to external data to be added to the chart, table or map, + folder_id: int, optional + New folder's ID + organization_id: str, optional + New organization's ID + forkable: bool, optional + Whether the chart should be forkable or not, by default None + language : str, optional + New language + metadata: dict, optional + A Python dictionary of properties to add. + + Return + ------ + dict + A dictionary containing the updated chart's information. + + Raises + ------ + InvalidRequest + If no updates are submitted. + """ + # Load the query with the provided parameters + _query: dict[str, Any] = {} + if title: + _query["title"] = title + if chart_type: + _query["type"] = chart_type + if theme: + _query["theme"] = theme + if external_data_url: + _query["externalData"] = external_data_url + if folder_id: + _query["folderId"] = folder_id + if organization_id: + _query["organizationId"] = organization_id + if forkable: + _query["forkable"] = json.dumps(forkable) + if language: + _query["language"] = language + if metadata: + _query["metadata"] = metadata + + # If there's nothing there to update, raise an exception + if not _query and not data: + msg = "No updates submitted." + logger.error(msg) + raise InvalidRequest(msg) + + # Update the chart + if _query: + obj = self.patch( + f"{self._CHARTS_URL}/{chart_id}", + data=_query, + extra_headers={"content-type": "application/json"}, + ) + else: + obj = self.get_chart(chart_id) + + # Add data, if provided + if data: + self.add_data(chart_id=obj["id"], data=data) + + # Return the result + return obj - # Add data if provided - if data is not None: - self.add_data(chart_id=chart_info["id"], data=data) + def update_metadata(self, chart_id: str, metadata: dict) -> dict: + """A deprecated method of the update_chart method.""" + # Issue a deprecation warning + logger.warning( + "This method is deprecated and will be removed in a future version. " + "Use update_chart instead." + ) - return chart_info + # Use the newer method + return self.update_chart(chart_id, metadata=metadata) def update_description( self, chart_id: str, - source_name: str = "", - source_url: str = "", - intro: str = "", - byline: str = "", - aria_description: str = "", - number_prepend: str = "", - number_append: str = "", - number_format: str = "-", - number_divisor: int = 0, + source_name: str | None = None, + source_url: str | None = None, + intro: str | None = None, + byline: str | None = None, + aria_description: str | None = None, + number_prepend: str | None = None, + number_append: str | None = None, + number_format: str | None = None, + number_divisor: int | None = None, ) -> dict: - """Update a chart's description. + """Update a chart's description attributes + + A convienece method for updating the 'describe' key of a chart's metadata. Parameters ---------- chart_id : str ID of chart, table or map. source_name : str, optional - Source of data, by default "" + Source of data source_url : str, optional - URL of source of data, by default "" + URL of source of data intro : str, optional - Introduction of your chart, table or map, by default "" + Introduction of your chart, table or map byline : str, optional - Who made this?, by default "" + Who made this? aria_description : str, optional Alt text description number_prepend : str, optional @@ -670,55 +806,82 @@ def update_description( The format number number_divisor : str, optional A multiplier or divisor for the numbers + + Returns + ------- + dict + A dictionary containing the updated chart's information. + + Raises + ------ + InvalidRequest + If no updates are submitted. """ - _query = { - "metadata": { - "describe": { - "source-name": source_name, - "source-url": source_url, - "intro": intro, - "byline": byline, - "aria-description": aria_description, - "number-prepend": number_prepend, - "number-append": number_append, - "number-format": number_format, - "number-divisor": number_divisor, - } - } - } - return self.patch( - f"{self._CHARTS_URL}/{chart_id}", - data=_query, - ) + # Load the query with the provided parameters + _query: dict[str, Any] = {} + if source_name: + _query["source-name"] = source_name + if source_url: + _query["source-url"] = source_url + if intro: + _query["intro"] = intro + if byline: + _query["byline"] = byline + if aria_description: + _query["aria-description"] = aria_description + if number_prepend: + _query["number-prepend"] = number_prepend + if number_append: + _query["number-append"] = number_append + if number_format: + _query["number-format"] = number_format + if number_divisor: + _query["number-divisor"] = number_divisor + + # If there's nothing there to update, raise an exception + if not _query: + msg = "No updates submitted." + logger.error(msg) + raise InvalidRequest(msg) - def publish_chart(self, chart_id: str, display: bool = True) -> dict | HTML: - """Publishes a chart, table or map. + # Update the chart using the update_chart method + metadata = {"metadata": {"describe": _query}} + return self.update_chart(chart_id, metadata=metadata) + + def delete_chart(self, chart_id: str) -> bool: + """Deletes a chart, table or map. Parameters ---------- chart_id : str ID of chart, table or map. - display : bool, optional - Display the published chart as output in notebook cell, by default True Returns ------- - dict | HTML - Either a dictionary containing the published chart's information or an HTML - object displaying the chart. + bool + True if the chart was deleted successfully. """ - chart_info = self.post(f"{self._PUBLISH_URL}/{chart_id}/publish") - assert isinstance(chart_info, dict) - if display: - iframe_code = chart_info["data"]["metadata"]["publish"]["embed-codes"][ - "embed-method-iframe" - ] - return HTML(iframe_code) - else: - return chart_info + return self.delete(f"{self._CHARTS_URL}/{chart_id}") - def chart_properties(self, chart_id: str) -> dict: - """Retrieve information of a specific chart, table or map. + def display_chart(self, chart_id: str) -> HTML: + """Displays a datawrapper chart. + + Parameters + ---------- + chart_id : str + ID of chart, table, or map. + + Returns + ------- + IPython.display.HTML + HTML displaying the chart. + """ + obj = self.get_chart(chart_id) + iframe = obj["metadata"]["publish"]["embed-codes"]["embed-method-iframe"] + return HTML(iframe) + + def copy_chart(self, chart_id: str) -> dict: + """Copy one of your charts, tables, or maps and create a new editable copy. Parameters ---------- @@ -730,10 +893,12 @@ def chart_properties(self, chart_id: str) -> dict: dict A dictionary containing the information of the chart, table, or map. """ - return self.get(self._CHARTS_URL + f"/{chart_id}") + response = self.post(f"{self._CHARTS_URL}/{chart_id}/copy") + assert isinstance(response, dict) + return response - def chart_data(self, chart_id: str): - """Retrieve the data stored for a specific chart, table or map, which is typically CSV. + def fork_chart(self, chart_id: str) -> dict: + """Fork a chart, table, or map and create an editable copy. Parameters ---------- @@ -745,139 +910,52 @@ def chart_data(self, chart_id: str): dict A dictionary containing the information of the chart, table, or map. """ - return self.get(self._CHARTS_URL + f"/{chart_id}/data") - - def update_metadata(self, chart_id: str, properties: dict) -> dict: - """Update a chart, table, or map's metadata. + response = self.post(f"{self._CHARTS_URL}/{chart_id}/fork") + assert isinstance(response, dict) + return response - Example: https://developer.datawrapper.de/docs/creating-a-chart-new#edit-colors + def publish_chart(self, chart_id: str, display: bool = False) -> dict | HTML: + """Publishes a chart, table or map. Parameters ---------- chart_id : str - ID of chart, table, or map. - properties : dict - A python dictionary of properties to update. + ID of chart, table or map. + display : bool, optional + Display the published chart as output in notebook cell, by default False + + Returns + ------- + dict | HTML + Either a dictionary containing the published chart's information or an HTML + object displaying the chart. """ - return self.patch( - f"{self._CHARTS_URL}/{chart_id}", - data={"metadata": properties}, - ) + obj = self.post(f"{self._CHARTS_URL}/{chart_id}/publish") + assert isinstance(obj, dict) + if display: + iframe = obj["data"]["metadata"]["publish"]["embed-codes"][ + "embed-method-iframe" + ] + return HTML(iframe) + else: + return obj - def update_chart( + def export_chart( self, chart_id: str, - title: str = "", - theme: str = "", - chart_type: str = "", - language: str = "", - folder_id: str = "", - organization_id: str = "", - ) -> dict | HTML: - """Updates a chart's title, theme, type, language, folder or organization. - - Parameters - ---------- - chart_id : str - ID Of chart, table, or map. - title : str, optional - New title, by default "" - theme : str, optional - New theme, by default "" - chart_type : str, optional - New chart type. See https://developer.datawrapper.de/docs/chart-types, - by default "" - language : str, optional - New language, by default "" - folder_id : str, optional - New folder's ID, by default "" - organization_id : str, optional - New organization's ID, by default "" - """ - _query = {} - if title: - _query["title"] = title - if theme: - _query["theme"] = theme - if chart_type: - _query["type"] = chart_type - if language: - _query["language"] = language - if folder_id: - _query["folderId"] = folder_id - if organization_id: - _query["organizationId"] = organization_id - - self.patch( - f"{self._CHARTS_URL}/{chart_id}", - data=_query, - ) - - return self.publish_chart(chart_id) - - def display_chart(self, chart_id: str) -> IPython.display.HTML: - """Displays a datawrapper chart. - - Parameters - ---------- - chart_id : str - ID of chart, table, or map. - - Returns - ------- - IPython.display.HTML - HTML displaying the chart. - """ - _chart_properties = self.chart_properties(chart_id) - _iframe_code = _chart_properties["metadata"]["publish"]["embed-codes"][ # type: ignore - "embed-method-iframe" - ] - - return HTML(_iframe_code) - - def get_iframe_code(self, chart_id: str, responsive: bool = False) -> str | Any: - """Returns a chart, table, or map's iframe embed code. - - Parameters - ---------- - chart_id : str - ID of chart, table, or map. - responsive : bool, optional - Whether to return a responsive iframe embed code., by default False - - Returns - ------- - str - iframe embed code. - """ - _chart_properties = self.chart_properties(chart_id) - - if responsive: - iframe_code = _chart_properties["metadata"]["publish"][ # type: ignore - "embed-codes" - ]["embed-method-responsive"] - else: - iframe_code = _chart_properties["metadata"]["publish"][ # type: ignore - "embed-codes" - ]["embed-method-iframe"] - return iframe_code - - def export_chart( - self, - chart_id: str, - unit: str = "px", - mode: str = "rgb", - width: int = 400, - plain: bool = False, - zoom: int = 2, - scale: int = 1, - border_width: int = 20, - transparent: bool = False, - output: str = "png", - filepath: str = "./image.png", - display: bool = False, - ) -> Any | None: - """Exports a chart, table, or map. + unit: str = "px", + mode: str = "rgb", + width: int = 400, + plain: bool = False, + zoom: int = 2, + scale: int = 1, + border_width: int = 20, + transparent: bool = False, + output: str = "png", + filepath: str = "./image.png", + display: bool = False, + ) -> Path | Image: + """Exports a chart, table, or map. Parameters ---------- @@ -911,101 +989,350 @@ def export_chart( Whether to display the exported image as output in the notebook cell, by default False - Returns None + Returns ------- - IPython.display.Image - If display is True, it returns an Image. + Path | Image + The file path to the exported image or an Image object displaying the image. """ - _export_url = f"{self._CHARTS_URL}/{chart_id}/export/{output}" - _filepath = Path(filepath) - _filepath = _filepath.with_suffix(f".{output}") - - _plain = "true" if plain else "false" - querystring = { + _query = { "unit": unit, "mode": mode, "width": width, - "plain": _plain, + "plain": json.dumps(plain), "zoom": zoom, "scale": scale, "borderWidth": border_width, "transparent": transparent, } - _header = self._auth_header - _header["accept"] = "*/*" + response = self.get( + f"{self._CHARTS_URL}/{chart_id}/export/{output}", params=_query + ) + + # Set the file path + _filepath = Path(filepath) + _filepath = _filepath.with_suffix(f".{output}") + + # Write the file to the file path + with open(_filepath, "wb") as fh: + fh.write(response.content) - export_chart_response = r.get( - url=_export_url, headers=_header, params=querystring # type: ignore + # Display the image if requested + if display: + return Image(_filepath) + # Otherwise return the file path + else: + logger.debug(f"File exported at {_filepath}") + return _filepath + + def get_iframe_code(self, chart_id: str, responsive: bool = False) -> str: + """Returns a chart, table, or map's iframe embed code. + + Parameters + ---------- + chart_id : str + ID of chart, table, or map. + responsive : bool, optional + Whether to return a responsive iframe embed code., by default False + + Returns + ------- + str + iframe embed code. + """ + obj = self.get_chart(chart_id) + if responsive: + iframe = obj["metadata"]["publish"]["embed-codes"][ + "embed-method-responsive" + ] + else: + iframe = obj["metadata"]["publish"]["embed-codes"]["embed-method-iframe"] + return iframe + + def get_data(self, chart_id: str): + """Retrieve the data stored for a specific chart, table or map, which is typically CSV. + + Parameters + ---------- + chart_id : str + ID of chart, table, or map. + + Returns + ------- + dict + A dictionary containing the information of the chart, table, or map. + """ + return self.get(self._CHARTS_URL + f"/{chart_id}/data") + + def chart_data(self, chart_id: str): + """A deprecated method of the get_data method.""" + # Issue a deprecation warning + logger.warning( + "This method is deprecated and will be removed in a future version. " + "Use get_data instead." ) - if export_chart_response.status_code == 200: - with open(_filepath, "wb") as response: - response.write(export_chart_response.content) - if display: - return Image(_filepath) - else: - logger.debug(f"File exported at {_filepath}") - return None - elif export_chart_response.status_code == 403: - msg = "You don't have access to the requested chart." + # Use the newer method + return self.get_data(chart_id) + + def add_data(self, chart_id: str, data: pd.DataFrame | str) -> bool: + """Add data to a specified chart. + + Parameters + ---------- + chart_id : str + ID of chart, table or map to add data to. + data : pd.DataFrame | str + A pandas dataframe containing the data to be added or a string that contains + the data. + + Returns + ------- + bool + True if the data was added successfully. + """ + # If data is a pandas dataframe, convert to csv + if isinstance(data, pd.DataFrame): + _data = data.to_csv(index=False, encoding="utf-8") + # If data is a string, use that + else: + _data = data + + # Add data to chart + return self.put( + f"{self._CHARTS_URL}/{chart_id}/data", + data=_data.encode("utf-8"), + dump_data=False, + ) + + def refresh_data(self, chart_id: str) -> dict: + """Fetch configured external data and add it to the chart. + + Parameters + ---------- + chart_id : str + ID of chart, table or map to add data to. + + Returns + ------- + dict + A dictionary containing the chart's information. + """ + response = self.post(f"{self._CHARTS_URL}/{chart_id}/data/refresh") + assert isinstance(response, dict) + return response + + def account_info(self) -> dict: + """A deprecated method for calling get_my_account.""" + # Issue a deprecation warning + logger.warning( + "This method is deprecated and will be removed in a future version. " + "Use get_account_info instead." + ) + + # Use the newer method + return self.get_my_account() + + def get_my_account(self) -> dict: + """Access your account information. + + Returns + ------- + dict + A dictionary containing your account information. + """ + return self.get(self._ME_URL) + + def update_my_account( + self, + name: str | None = None, + email: str | None = None, + role: str | None = None, + language: str | None = None, + password: str | None = None, + old_password: str | None = None, + ) -> dict: + """Update your account information. + + Parameters + ---------- + name : str, optional + Your new name, by default None + email : str, optional + Your new email, by default None + role : str, optional + Your new role, by default None + language: str, optional + Your new language, by default None + password: str, optional + Your new, strong password, by default None + old_password: str, optional + Your previous password, by default None + + Returns + ------- + dict + A dictionary containing your updated account information. + """ + _query: dict = {} + if name: + _query["name"] = name + if email: + _query["email"] = email + if role: + _query["role"] = role + if language: + _query["language"] = language + if password and old_password: + _query["password"] = password + _query["oldPassword"] = old_password + if password and not old_password: + msg = "You must provide your old password to change it." logger.error(msg) raise Exception(msg) - elif export_chart_response.status_code == 401: - msg = "You couldn't be authenticated." + if old_password and not password: + msg = "You must provide a new password to change it." logger.error(msg) raise Exception(msg) - else: - msg = "Chart could not be exported." + + return self.patch( + self._ME_URL, + data=_query, + ) + + def update_my_settings( + self, + active_team: str | None = None, + ) -> dict: + """Update your account information. + + Parameters + ---------- + active_team: str, optional + Your active team + + Returns + ------- + dict + The user settings dictionary following the change. + """ + _query: dict = {} + if active_team: + _query["activeTeam"] = active_team + + if not _query: + msg = "No updates submitted." logger.error(msg) raise Exception(msg) - def get_basemaps(self) -> list[dict]: - """Get a list of the available basemaps. + return self.patch( + f"{self._ME_URL}/settings", + data=_query, + ) + + def get_my_recently_edited_charts( + self, + limit: str | int = 100, + offset: str | int = 0, + min_last_edit_step: str | int = 0, + ) -> dict: + """Get a list of your recently edited charts. + + Parameters + ---------- + limit: str | int + Maximum items to fetch. Useful for pagination. 100 by default. + offset: str | int + Number of items to skip. Useful for pagination. Zero by default. + min_last_edit_step: str | int + Filter visualizations by the last editor step they've + been opened in (1=upload, 2=describe, 3=visualize, etc). + Zero by default. Returns ------- - list[dict] - A list of dictionaries containing the basemaps available in your Datawrapper account. + dict + A dictionary with the list of charts and metadata about the selection. """ - return self.get(self._BASEMAPS_URL) + _query: dict = {} + if limit: + _query["limit"] = limit + if offset: + _query["offset"] = offset + if min_last_edit_step: + _query["minLastEditStep"] = min_last_edit_step - def get_basemap(self, basemap_id: str, wgs84: bool = False) -> dict: - """Get the metdata of the requested basemap. + return self.get( + self._ME_URL + "/recently-edited-charts", + params=_query, + ) + + def get_my_recently_published_charts( + self, + limit: str | int = 100, + offset: str | int = 0, + min_last_edit_step: str | int = 0, + ) -> dict: + """Get a list of your recently published charts. Parameters ---------- - basemap_id : str - ID of basemap to get. - wgs84 : bool, optional - Whether to return the basemap in the WGS84 project, by default False + limit: str | int + Maximum items to fetch. Useful for pagination. 100 by default. + offset: str | int + Number of items to skip. Useful for pagination. Zero by default. + min_last_edit_step: str | int + Filter visualizations by the last editor step they've + been opened in (1=upload, 2=describe, 3=visualize, etc). + Zero by default. Returns ------- dict - A dictionary containing the requested basemap's metadata. + A dictionary with the list of charts and metadata about the selection. """ + _query: dict = {} + if limit: + _query["limit"] = limit + if offset: + _query["offset"] = offset + if min_last_edit_step: + _query["minLastEditStep"] = min_last_edit_step + return self.get( - f"{self._BASEMAPS_URL}/{basemap_id}", - params={"wgs84": wgs84}, + self._ME_URL + "/recently-published-charts", + params=_query, ) - def get_basemap_key(self, basemap_id: str, basemap_key: str) -> dict: - """Get the list of available values for a basemap's key. + def get_themes( + self, limit: str | int = 100, offset: str | int = 0, deleted: bool = False + ) -> dict: + """Get a list of themes in your Datawrapper account. Parameters ---------- - basemap_id : str - ID of basemap to get. - basemap_key : str - Metadata key of basemap to get. + limit: str | int + Maximum items to fetch. Useful for pagination. Default 100. + offset: str | int + Number of items to skip. Useful for pagination. Default zero. + deleted: bool + Whether to include deleted themes Returns ------- dict - A dictionary containing the requested data. + A dictionary containing the themes in your Datawrapper account. """ - return self.get(f"{self._BASEMAPS_URL}/{basemap_id}/{basemap_key}") + _query = { + "limit": limit, + "offset": offset, + "deleted": json.dumps(deleted), + } + + return self.get( + self._THEMES_URL, + params=_query, + ) def get_folders(self) -> dict: """Get a list of folders in your Datawrapper account. @@ -1143,55 +1470,6 @@ def move_chart(self, chart_id: str, folder_id: str) -> dict: data={"folderId": folder_id}, ) - def copy_chart(self, chart_id: str) -> dict: - """Copy one of your charts, tables, or maps and create a new editable copy. - - Parameters - ---------- - chart_id : str - ID of chart, table, or map. - - Returns - ------- - dict - A dictionary containing the information of the chart, table, or map. - """ - response = self.post(f"{self._CHARTS_URL}/{chart_id}/copy") - assert isinstance(response, dict) - return response - - def fork_chart(self, chart_id: str) -> dict: - """Fork a chart, table, or map and create an editable copy. - - Parameters - ---------- - chart_id : str - ID of chart, table, or map. - - Returns - ------- - dict - A dictionary containing the information of the chart, table, or map. - """ - response = self.post(f"{self._CHARTS_URL}/{chart_id}/fork") - assert isinstance(response, dict) - return response - - def delete_chart(self, chart_id: str) -> bool: - """Deletes a specified chart, table or map. - - Parameters - ---------- - chart_id : str - ID of chart, table, or map. - - Returns - ------- - bool - True if the chart was deleted successfully. - """ - return self.delete(f"{self._CHARTS_URL}/{chart_id}") - def get_charts( self, user_id: str = "", @@ -1251,188 +1529,6 @@ def get_charts( return self.get(self._CHARTS_URL, params=_query) - def get_api_tokens(self, limit: int = 100, offset: int = 0) -> dict[str, Any]: - """Retrieves all API tokens associated to the current user. - - Response will not include full tokens for security reasons. Requires scope `auth:read`. - - Parameters - ---------- - limit : int, optional - Maximum items to fetch, by default 100. Useful for pagination. - offset : int, optional - Offset for pagination, by default 0. - - Returns - ------- - dict - A dictionary containing the API tokens for your Datawrapper account. - """ - _query: dict[str, Any] = {} - if limit: - _query["limit"] = limit - if offset: - _query["offset"] = offset - - return self.get(self._API_TOKEN_URL, params=_query) - - def create_api_token(self, comment: str, scopes: list) -> dict: - """Create a new API Token. - - Make sure to save the token somewhere, since you won't be able to see it again. Requires scope `auth:write`. - - Parameters - ---------- - comment : str - Comment to describe the API token. Tip: Use something to remember where this specific token is used. - scopes : list - List of scopes for the API token. - - Returns - ------- - dict - A dictionary containing the API token's information. - """ - response = self.post( - self._API_TOKEN_URL, - data={"comment": comment, "scopes": scopes}, - extra_headers={"content-type": "application/json"}, - ) - assert isinstance(response, dict) - return response - - def update_api_token( - self, id: str | int, comment: str, scopes: list | None = None - ) -> bool: - """Updates an existing API token. - - Parameters - ---------- - id : str | int - ID of API token to update. - comment : str - Comment to describe the API token. Tip: Use something to remember where this specific token is used. - scopes : list, optional - List of scopes for the API token. - - Returns - ------- - bool - True if the API token was updated successfully. - """ - _query: dict = {"comment": comment} - if scopes: - _query["scopes"] = scopes - - return self.put( - f"{self._API_TOKEN_URL}/{id}", - data=_query, - extra_headers={"content-type": "application/json"}, - ) - - def delete_api_token(self, token_id: str | int) -> bool: - """Deletes an API token. - - Parameters - ---------- - token_id : str | int - ID of API token to delete. - - Returns - ------- - bool - True if the API token was deleted successfully. - """ - return self.delete(f"{self._API_TOKEN_URL}/{token_id}") - - def get_login_tokens( - self, - limit: int = 100, - offset: int = 0, - ) -> dict[str, Any]: - """Retrieves all login tokens associated to the current user. - - Parameters - ---------- - limit : int, optional - Maximum items to fetch, by default 100. Useful for pagination. - offset : int, optional - Offset for pagination, by default 0. - - Returns - ------- - dict - A dictionary containing the login tokens for your Datawrapper account. - """ - _query: dict[str, Any] = {} - if limit: - _query["limit"] = limit - if offset: - _query["offset"] = offset - - return self.get(self._LOGIN_TOKENS_URL, params=_query) - - def create_login_token( - self, - ) -> dict: - """Creates a new login token to authenticate a user, for use in CMS integrations. - - Login tokens are valid for five minutes and can only be used once. - - Returns - ------- - dict - A dictionary containing the login token's information. - """ - response = self.post( - self._LOGIN_TOKENS_URL, - extra_headers={"content-type": "application/json"}, - ) - assert isinstance(response, dict) - return response - - def delete_login_token(self, token_id: str | int) -> bool: - """Deletes a login token. - - Parameters - ---------- - token_id : str | int - ID of login token to delete. - - Returns - ------- - bool - True if the login token was deleted successfully. - """ - return self.delete(f"{self._LOGIN_TOKENS_URL}/{token_id}") - - def login(self, token: str) -> str: - """Login using a one-time login token and redirect to the URL associated with the token. - - For use in CMS integrations. - - Parameters - ---------- - token : str - Login token. - - Returns - ------- - str - The HTML of the page that the token redirects to. - """ - return self.get(f"{self._LOGIN_URL}/{token}") - - def get_token_scopes(self) -> list: - """Get the scopes that are available to the current user. - - Returns - ------- - list - A list containing the scopes available to the current user. - """ - return self.get(self._LOGIN_SCOPES_URL) - def get_teams( self, search: str | None = None, diff --git a/datawrapper/exceptions.py b/datawrapper/exceptions.py index d54e0d4..46fb508 100644 --- a/datawrapper/exceptions.py +++ b/datawrapper/exceptions.py @@ -8,3 +8,11 @@ def __init__(self, response): """Initialize the exception.""" msg = f"Request failed with status code {response.status_code}. Response content: {response.content}" super().__init__(msg) + + +class InvalidRequest(Exception): + """Custom exception for invalid API requests.""" + + def __init__(self, message): + """Initialize the exception.""" + super().__init__(message) diff --git a/setup.py b/setup.py index 2cad623..e7579ea 100644 --- a/setup.py +++ b/setup.py @@ -47,7 +47,7 @@ def local_version(version): setup( name="datawrapper", - description="A light-weight python wrapper for the Datawrapper API", + description="A lightweight Python wrapper for the Datawrapper API", long_description=read("README.md"), long_description_content_type="text/markdown", author="chekos",