Skip to content

notion_recurring_tasks.notion_api

Code related to the interaction with the Notion API.

Calls

Gathers the available API calls. All methods return the response in json-format and raises an HTTPError if unsuccesful.

db_query staticmethod 

db_query(db_id: str, params: dict) -> dict

Query a Notion database with a given id.

Parameters:

Name Type Description Default
db_id str

Notion database id.

 required
params dict

Input parameters for the query.

 required

Raises:

Type Description
HTTPError

If API call is unsuccessful.

Returns:

Type Description
dict

Response data in json format.
Source code in src/notion_recurring_tasks/notion_api.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@staticmethod
def db_query(db_id: str, params: dict) -> dict:
    """Query a Notion database with a given id.

    Args:
        db_id (str): Notion database id.
        params (dict): Input parameters for the query.

    Raises:
        HTTPError: If API call is unsuccessful.

    Returns:
        dict: Response data in json format.
    """
    response = requests.post(
        f"{NOTION_API_URL}/databases/{db_id}/query",
        headers=get_headers(),
        json=params,
    )
    response.raise_for_status()
    return response.json()

patch_page_properties staticmethod 

patch_page_properties(
    page_id: str, new_properties: dict
) -> dict

Patch (update) a page with new properties.

Parameters:

Name Type Description Default
page_id str

Notion page id

 required
new_properties dict

New properties to add to page

 required

Raises:

Type Description
HTTPError

If API call is unsuccessful

Returns:

Type Description
dict

Response data in json format
Source code in src/notion_recurring_tasks/notion_api.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@staticmethod
def patch_page_properties(page_id: str, new_properties: dict) -> dict:
    """Patch (update) a page with new properties.

    Args:
        page_id (str): Notion page id
        new_properties (dict): New properties to add to page

    Raises:
        HTTPError: If API call is unsuccessful

    Returns:
        dict: Response data in json format
    """
    properties = {"properties": new_properties}
    response = requests.patch(
        f"{NOTION_API_URL}/pages/{page_id}", headers=get_headers(), json=properties
    )
    response.raise_for_status()
    return response.json()

search staticmethod 

search(params: dict) -> dict

Performs a search query with the given params.

Parameters:

Name Type Description Default
params dict

Input parameters for the search query.

 required

Raises:

Type Description
HTTPError

If API call is unsuccessful.

Returns:

Type Description
dict

Response data in json format.
Source code in src/notion_recurring_tasks/notion_api.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@staticmethod
def search(params: dict) -> dict:
    """Performs a search query with the given params.

    Args:
        params (dict): Input parameters for the search query.

    Raises:
        HTTPError: If API call is unsuccessful.

    Returns:
        dict: Response data in json format.
    """
    response = requests.post(
        f"{NOTION_API_URL}/search", headers=get_headers(), json=params
    )
    response.raise_for_status()
    return response.json()

find_recurrable_tasks 

find_recurrable_tasks(db_id: str) -> list

Finds all tasks that are due to recur from the database.

Parameters:

Name Type Description Default
db_id str

The id of the database to be queried.

 required

Returns:

Type Description
list

List of all tasks found.
Source code in src/notion_recurring_tasks/notion_api.py 
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
145
146
147
148
def find_recurrable_tasks(db_id: str) -> list:
    """Finds all tasks that are due to recur from the database.

    Args:
        db_id (str): The id of the database to be queried.

    Returns:
        list: List of all tasks found.
    """
    TASK_FILTERS: dict = {
        "filter": {
            "and": [
                {
                    "property": "Recurring",
                    "checkbox": {"equals": True},
                },
                {
                    "property": "Status",
                    "status": {"equals": "Done"},
                },
                {
                    "property": "Archived",
                    "checkbox": {"equals": False},
                },
            ]
        },
    }

    data = Calls.db_query(db_id, TASK_FILTERS)

    return data["results"]

get_db_id 

get_db_id() -> str

Gets id of Tasks database.

Raises:

Type Description
ValueError

If database id was not found.

Returns:

Type Description
str

The database id
Source code in src/notion_recurring_tasks/notion_api.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def get_db_id() -> str:
    """Gets id of Tasks database.

    Raises:
        ValueError: If database id was not found.

    Returns:
        str: The database id
    """
    SEARCH_PARAMS: dict = {"filter": {"value": "database", "property": "object"}}

    results: list[dict] = Calls.search(SEARCH_PARAMS)["results"]

    if results[0]["object"] != "database":
        # User may have not created the connection in the database
        raise ValueError("Database not found..")

    return results[0]["id"]

get_headers 

get_headers() -> dict[str, str]

Get the headers for the Notion API call.

Returns:

Type Description
dict[str, str]

Header dictionary for Notion API
Source code in src/notion_recurring_tasks/notion_api.py 
16
17
18
19
20
21
22
23
24
25
26
def get_headers() -> dict[str, str]:
    """Get the headers for the Notion API call.

    Returns:
        dict[str, str]: Header dictionary for Notion API
    """
    return {
        "Authorization": f"Bearer {os.getenv('NOTION_KEY')}",
        "Content-Type": "application/json",
        "Notion-Version": "2022-06-28",
    }

update_task_properties 

update_task_properties(
    task_id: str, new_due_date: str | None
) -> None

Reset the task and set a new due date if provided.

Parameters:

Name Type Description Default
task_id str

The id of the task

 required
new_due_date str | None

The new due date to set in ISO-format 'YYYY-MM-DD'

 required
Source code in src/notion_recurring_tasks/notion_api.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def update_task_properties(task_id: str, new_due_date: str | None) -> None:
    """Reset the task and set a new due date if provided.

    Args:
        task_id (str): The id of the task
        new_due_date (str | None): The new due date to set in ISO-format 'YYYY-MM-DD'
    """
    new_properties: dict = {}

    ## Set due date
    if new_due_date:
        new_properties["Due Date"] = {
            "date": {
                "start": new_due_date,
            }
        }
    else:
        new_properties["Due Date"] = {
            "date": None,
        }

    ## Set status
    new_properties["Status"] = {
        "status": {
            "name": "Not started",
        }
    }

    Calls.patch_page_properties(task_id, new_properties)