notion

Module diagram

classDiagram
  class notion {
  }
  class block {
  }
  class common {
  }
  class database {
  }
  class filter {
  }
  class property {
  }
  class user {
  }
  notion --> block
  notion --> database
  notion --> user
  block --> common
  database --> common
  database --> property
  property --> common
  property --> filter
  property --> user
  user --> common

notion

retrieve_block_children

retrieve_block_children(page_id: str, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None)

get all blocks on notion page. Args: page_id (str): notion page_id to search token_path (str, optional): description. Defaults to RESEARCH_API_PATH. token (Optional[str], optional): description. Defaults to None.

Returns:

• _type_ –

  request response code (200 if succeeded) and response text, block_children

Source code in ResearchToolkit/notion/block.py

def retrieve_block_children(
    page_id: str,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
):
    """
    get all blocks on notion page.
    Args:
        page_id (str): notion page_id to search
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.

    Returns:
        _type_: request response code (200 if succeeded) and response text, block_children
    """
    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
    }

    api_url = f"https://api.notion.com/v1/blocks/{page_id}/children?page_size=100"

    # get block children from notion
    response = requests.get(api_url, headers=headers)

    try:
        block_children = response.json()["results"]
    except AttributeError:
        print("failed to get block children")
        block_children = None

    return response.status_code, response.text, block_children

find_database_on_page

find_database_on_page(page_id: str) -> Dict[str, str]

find all database on page.

Returns:

• Dict[str, str] –

  Dict[str, str]: dictionary whose keys are database ids and values are database names

Source code in ResearchToolkit/notion/block.py

def find_database_on_page(page_id: str) -> Dict[str, str]:
    """
    find all database on page.

    Returns:
        Dict[str, str]: dictionary whose keys are database ids and values are database names

    """
    _, _, block_children = retrieve_block_children(page_id)

    databases = dict()
    if block_children is None:
        return databases
    for block in block_children:
        if block["type"] == "child_database":
            database = {block["id"]: block["child_database"]["title"]}
            databases.update(database)
    return databases

append_block_children

append_block_children(page_id: str, block: Union[dict, List[dict]], token_path: str = RESEARCH_API_PATH, token: Optional[str] = None)

append block or blocks to a page

Parameters:

• page_id (str) –

  description

• block (Union[dict, List[dict]]) –

  output or list of outputs of Block objects e.g) Paragraph("hello world")() or [Heading1("Tutorial")(), Paragraph("hello world")()]

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

Returns:

• _type_ –

  request response code (200 if succeeded) and response text

Source code in ResearchToolkit/notion/block.py

def append_block_children(
    page_id: str,
    block: Union[dict, List[dict]],
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
):
    """append block or blocks to a page

    Args:
        page_id (str): _description_
        block (Union[dict, List[dict]]): output or list of outputs of `Block` objects
                                         e.g) Paragraph("hello world")() or [Heading1("Tutorial")(), Paragraph("hello world")()]
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.

    Returns:
        _type_: request response code (200 if succeeded) and response text
    """
    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Notion-Version": "2022-06-28",
    }

    api_url = f"https://api.notion.com/v1/blocks/{page_id}/children"

    if isinstance(block, dict):
        block = [block]
    data = {"children": block}
    data = json.dumps(data)

    response = requests.request("PATCH", api_url, headers=headers, data=data)

    return response.status_code, response.text

query_database

query_database(database_id: str, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None, request_body: Optional[dict] = None)

query a database.
refer https://developers.notion.com/reference/post-database-query

Parameters:

• database_id (str) –

  description

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

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

  request body for query with filtering or sorting. Defaults to None. Check https://developers.notion.com/reference/post-database-query for detailed usage of BODY_PARAM.

Returns:

• _type_ –

  request response code (200 if succeeded), response text, results (None if request failed)

Source code in ResearchToolkit/notion/database.py

def query_database(
    database_id: str,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
    request_body: Optional[dict] = None,
):
    """
        query a database.
        refer https://developers.notion.com/reference/post-database-query

    Args:
        database_id (str): _description_
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.
        request_body (Optional[dict], optional): request body for query with filtering or sorting. Defaults to None.
            Check https://developers.notion.com/reference/post-database-query for detailed usage of BODY_PARAM.

    Returns:
        _type_: request response code (200 if succeeded), response text, results (None if request failed)
    """

    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
        "accept": "application/json",
        "content-type": "application/json",
    }

    api_url = f"https://api.notion.com/v1/databases/{database_id}/query"

    request_body = {
        "page_size": QUERY_PAGE_SIZE,
        **(request_body or {}),
    }

    results = None
    while True:
        response = requests.post(api_url, headers=headers, json=request_body)
        if response.status_code != RESPONSE_SUCCESS:
            raise RuntimeError(f"Notion Error: {response.text}")

        pages = response.json()["results"]
        if results is None:
            results = pages
        else:
            assert pages[0]["id"] == results[-1]["id"]
            results += pages[1:]

        if len(pages) < QUERY_PAGE_SIZE:
            break

        request_body["start_cursor"] = pages[-1]["id"]

    return response.status_code, response.text, results

create_database

create_database(page_id: str, fields: FieldType, relations: Dict[str, Dict[str, str]] = {}, title: Optional[str] = None, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None)

creates a notion database in a page_id page. it is recommended to add "user" field in fields which specifies the user of api when collaborating on a database.

Parameters:

• page_id (str) –

  notion page id

• fields (FieldType) –

  key: name of the property, value: property type (should be one of the keys in PROPERTIES) e.g) {"exp_name": "title", "accuracy": "number"} create a database whose property names are exp_name, accuracy and their property types are title and number.

• relations (Dict[str, Dict[str, str]], default: {} ) –

  key: name of the relation property, value: dictionary with "database_id" and "type" info. "database_id" is notion database id to relate to "type" should be one of "single_property" or "dual_property". e.g) {"related_task": {"database_id": (database_id), "type": "dual_property}} !NOTE: relation database should also be shared with notion integration!

• title (str, default: None ) –

  title of database. Defaults to None ("untitled" database will be created).

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

Returns:

• _type_ –

  request response code (200 if succeeded), response text, database_id (None if request failed)

Source code in ResearchToolkit/notion/database.py

def create_database(
    page_id: str,
    fields: FieldType,
    relations: Dict[str, Dict[str, str]] = {},
    title: Optional[str] = None,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
):
    """creates a notion database in a page_id page.
       it is recommended to add "user" field in `fields` which specifies the user of api when collaborating on a database.

    Args:
        page_id (str): notion page id
        fields (FieldType): key: name of the property, value: property type (should be one of the keys in `PROPERTIES`)
                                 e.g) {"exp_name": "title", "accuracy": "number"}
                                      create a database whose property names are exp_name, accuracy and their property types are title and number.
        relations (Dict[str, Dict[str, str]]): key: name of the relation property, value: dictionary with "database_id" and "type" info.
                                               "database_id" is notion database id to relate to
                                               "type" should be one of "single_property" or "dual_property".
                                               e.g) {"related_task": {"database_id": (database_id), "type": "dual_property}}
                                               !NOTE: relation database should also be shared with notion integration!
        title (str, optional): title of database. Defaults to None ("untitled" database will be created).
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.

    Returns:
        _type_: request response code (200 if succeeded), response text, database_id (None if request failed)
    """
    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Notion-Version": "2022-06-28",
    }

    api_url = "https://api.notion.com/v1/databases"

    # define a database
    property_types = fields.values()
    assert "title" in property_types, "title should be in properties"

    properties = make_database_property_schema(fields=fields, relations=relations)

    data = {"parent": {"type": "page_id", "page_id": page_id}, "properties": properties}

    if title:
        data.update(Title(title)())

    data = json.dumps(data)

    # create database in notion
    response = requests.request("POST", api_url, headers=headers, data=data)

    try:
        database_id = response.json()["id"]
    except AttributeError:
        print("database creation failed")
        database_id = None

    return response.status_code, response.text, database_id

retrieve_database_fields

retrieve_database_fields(database_id: str, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None, is_page: bool = False)

retrieves database fields

Parameters:

• database_id (_type_) –

  notion database id

• token_path (_type_, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (_type_, default: None ) –

  description. Defaults to None.

• is_page (bool, default: False ) –

  True if the id is for a page. Defaults to None.

Returns:

• _type_ –

  fields dictionary - key: property name, value: property type e.g) {"exp_name": "title", "accuracy": "number"}

Source code in ResearchToolkit/notion/database.py

def retrieve_database_fields(
    database_id: str,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
    is_page: bool = False,
):
    """retrieves database fields

    Args:
        database_id (_type_): notion database id
        token_path (_type_, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (_type_, optional): _description_. Defaults to None.
        is_page (bool): True if the id is for a page. Defaults to None.

    Returns:
        _type_: fields dictionary - key: property name, value: property type
                                      e.g) {"exp_name": "title", "accuracy": "number"}
    """
    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
    }

    sub_url = "pages" if is_page else "databases"
    api_url = f"https://api.notion.com/v1/{sub_url}/{database_id}"

    # get database from notion
    response = requests.get(api_url, headers=headers)

    # extract database fields
    database_fields = {
        name: prop["type"]
        .replace("rich_text", "text")
        .replace("people", "person")
        .replace("phone_number", "phone")
        for name, prop in response.json()["properties"].items()
    }

    return database_fields

add_data_to_database

add_data_to_database(database_id: str, data: dict, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None, user: Optional[Dict[str, BasicProperty]] = None)

add data to notion database. properties of data values are assigned to match the properties of existing database.

when collaborating on a database, it is recommended to use user option to specfiy who edited the database.

Parameters:

• database_id (str) –

  description

• data (dict) –

  data to store in database. keys should match names of the database properties

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

• user (Optional[Dict[str, BasicProperty]], default: None ) –

                          use when collaborating on a databaese to specify editor of the database.
                          key should be the property name of the database responsible for tracking api user.
                          if vaule is `Person`, a notification will be send in notion.
  

Returns:

• _type_ –

  request response code (200 if succeeded) and response text, page_id (None if request failed)

Source code in ResearchToolkit/notion/database.py

def add_data_to_database(
    database_id: str,
    data: dict,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
    user: Optional[Dict[str, BasicProperty]] = None,
):
    """
    add data to notion database.
    properties of data values are assigned to match the properties of existing database.

    when collaborating on a database, it is recommended to use `user` option to specfiy who edited the database.

    Args:
        database_id (str): _description_
        data (dict): data to store in database. keys should match names of the database properties
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.
        user (Optional[Dict[str, BasicProperty]], optional):
                                        use when collaborating on a databaese to specify editor of the database.
                                        key should be the property name of the database responsible for tracking api user.
                                        if vaule is `Person`, a notification will be send in notion.

    Returns:
        _type_: request response code (200 if succeeded) and response text, page_id (None if request failed)
    """
    database_fields = retrieve_database_fields(database_id, token_path=token_path, token=token)

    properties = convert_data_into_property(data, database_fields)

    if user and list(user.keys())[0] in database_fields:
        properties.update(user)

    return _add_to_database(database_id, properties, token_path=token_path, token=token)

get_database_in_pd_dataframe

get_database_in_pd_dataframe(database_id: str, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None, request_body: Optional[dict] = None)

get notion database and return it in pandas dataframe.

Parameters:

• database_id (str) –

  description

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

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

  request body for query with filtering or sorting. Defaults to None. Check https://developers.notion.com/reference/post-database-query for detailed usage of BODY_PARAM.

Returns:

• –

  pd.DataFrame: description

Source code in ResearchToolkit/notion/database.py

def get_database_in_pd_dataframe(
    database_id: str,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
    request_body: Optional[dict] = None,
):
    """
    get notion database and return it in pandas dataframe.

    Args:
        database_id (str): _description_
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.
        request_body (Optional[dict], optional): request body for query with filtering or sorting. Defaults to None.
            Check https://developers.notion.com/reference/post-database-query for detailed usage of BODY_PARAM.

    Returns:
        pd.DataFrame: _description_
    """
    status, text, results = query_database(database_id, token_path, token, request_body)

    data = {key: [] for key in results[0]["properties"].keys()}
    for result in results:
        properties = result["properties"]
        for key, property_ in properties.items():
            value = get_property_value(property_)
            data[key].append(value)

    return pd.DataFrame(data)

get_notion_users

get_notion_users()

Source code in ResearchToolkit/notion/user.py

def get_notion_users():
    token = get_token(RESEARCH_API_PATH)

    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
    }

    api_url = "https://api.notion.com/v1/users"

    response = requests.get(api_url, headers=headers)
    result = response.json()["results"]

    users = {
        user["name"]: {"id": user["id"], "email": user["person"]["email"]}
        for user in result
        if "person" in user
    }

    return users

block

BLOCKS module-attribute

BLOCKS = {}

RICH_TEXT_LIMIT module-attribute

RICH_TEXT_LIMIT = 2000

Block

Block(value, **kwargs)

Notion Block objects. It converts values into block object structures specified by the notion api. Additionally, it can check whether given value is valid for the property.

Source code in ResearchToolkit/notion/block.py

def __init__(self, value, **kwargs):
    self.value = value
    self.out = {self.name: self.value}

value instance-attribute

value = value

out instance-attribute

out = {name: value}

__init_subclass__

__init_subclass__(**kwargs)

Source code in ResearchToolkit/notion/block.py

def __init_subclass__(cls, **kwargs):
    super().__init_subclass__(**kwargs)
    cls.name = pascal_to_snake(cls.__name__)
    register(cls)

__call__

__call__()

Source code in ResearchToolkit/notion/block.py

def __call__(self):
    return self.out

Heading1

Heading1(value, color: Optional[str] = None)

Bases: Block

Source code in ResearchToolkit/notion/block.py

def __init__(self, value, color: Optional[str] = None):
    super(Heading1, self).__init__(value)
    self.out = {self.name: {"rich_text": [{"text": {"content": value}}]}}
    if color in [
        "default",
        "gray",
        "brown",
        "orange",
        "yellow",
        "green",
        "blue",
        "purple",
        "pink",
        "red",
        "gray_background",
        "brown_background",
        "orange_background",
        "yellow_background",
        "green_background",
        "blue_background",
        "purple_background",
        "pink_background",
        "red_background",
    ]:
        self.out[self.name].update({"color": color})

out instance-attribute

out = {name: {'rich_text': [{'text': {'content': value}}]}}

__bool__

__bool__()

Source code in ResearchToolkit/notion/block.py

def __bool__(self):
    return isinstance(self.value, str)

Heading2

Heading2(value, color: Optional[str] = None)

Bases: Heading1

Source code in ResearchToolkit/notion/block.py

def __init__(self, value, color: Optional[str] = None):
    super(Heading1, self).__init__(value)
    self.out = {self.name: {"rich_text": [{"text": {"content": value}}]}}
    if color in [
        "default",
        "gray",
        "brown",
        "orange",
        "yellow",
        "green",
        "blue",
        "purple",
        "pink",
        "red",
        "gray_background",
        "brown_background",
        "orange_background",
        "yellow_background",
        "green_background",
        "blue_background",
        "purple_background",
        "pink_background",
        "red_background",
    ]:
        self.out[self.name].update({"color": color})

Heading3

Heading3(value, color: Optional[str] = None)

Bases: Heading1

Source code in ResearchToolkit/notion/block.py

def __init__(self, value, color: Optional[str] = None):
    super(Heading1, self).__init__(value)
    self.out = {self.name: {"rich_text": [{"text": {"content": value}}]}}
    if color in [
        "default",
        "gray",
        "brown",
        "orange",
        "yellow",
        "green",
        "blue",
        "purple",
        "pink",
        "red",
        "gray_background",
        "brown_background",
        "orange_background",
        "yellow_background",
        "green_background",
        "blue_background",
        "purple_background",
        "pink_background",
        "red_background",
    ]:
        self.out[self.name].update({"color": color})

Paragraph

Paragraph(value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None)

Bases: Heading1

summary

Parameters:

• value (_type_) –

  description

• color (Optional[str], default: None ) –

  description. Defaults to None.

• children (Optional[Union[dict, List[dict]]], default: None ) –

  children should be output or list of Block object output. e.g) Paragraph("hello world")() or [Heading1("Tutorial")(), Paragraph("hello world")()]

Source code in ResearchToolkit/notion/block.py

def __init__(
    self, value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None
):
    """_summary_

    Args:
        value (_type_): _description_
        color (Optional[str], optional): _description_. Defaults to None.
        children (Optional[Union[dict, List[dict]]], optional): children should be output or list of `Block` object output.
                                                                e.g) Paragraph("hello world")()
                                                                    or [Heading1("Tutorial")(), Paragraph("hello world")()]
    """
    super(Paragraph, self).__init__(value, color)
    if children:
        if isinstance(children, dict):
            children = [children]
        self.out[self.name].update({"children": children})

Callout

Callout(value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None)

Bases: Paragraph

summary

Parameters:

• value (_type_) –

  description

• color (Optional[str], default: None ) –

  description. Defaults to None.

• children (Optional[Union[dict, List[dict]]], default: None ) –

  children should be output or list of Block object output. e.g) Paragraph("hello world")() or [Heading1("Tutorial")(), Paragraph("hello world")()]

Source code in ResearchToolkit/notion/block.py

def __init__(
    self, value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None
):
    """_summary_

    Args:
        value (_type_): _description_
        color (Optional[str], optional): _description_. Defaults to None.
        children (Optional[Union[dict, List[dict]]], optional): children should be output or list of `Block` object output.
                                                                e.g) Paragraph("hello world")()
                                                                    or [Heading1("Tutorial")(), Paragraph("hello world")()]
    """
    super(Paragraph, self).__init__(value, color)
    if children:
        if isinstance(children, dict):
            children = [children]
        self.out[self.name].update({"children": children})

Quote

Quote(value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None)

Bases: Paragraph

summary

Parameters:

• value (_type_) –

  description

• color (Optional[str], default: None ) –

  description. Defaults to None.

• children (Optional[Union[dict, List[dict]]], default: None ) –

  children should be output or list of Block object output. e.g) Paragraph("hello world")() or [Heading1("Tutorial")(), Paragraph("hello world")()]

Source code in ResearchToolkit/notion/block.py

def __init__(
    self, value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None
):
    """_summary_

    Args:
        value (_type_): _description_
        color (Optional[str], optional): _description_. Defaults to None.
        children (Optional[Union[dict, List[dict]]], optional): children should be output or list of `Block` object output.
                                                                e.g) Paragraph("hello world")()
                                                                    or [Heading1("Tutorial")(), Paragraph("hello world")()]
    """
    super(Paragraph, self).__init__(value, color)
    if children:
        if isinstance(children, dict):
            children = [children]
        self.out[self.name].update({"children": children})

BulletedListItem

BulletedListItem(value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None)

Bases: Paragraph

summary

Parameters:

• value (_type_) –

  description

• color (Optional[str], default: None ) –

  description. Defaults to None.

• children (Optional[Union[dict, List[dict]]], default: None ) –

  children should be output or list of Block object output. e.g) Paragraph("hello world")() or [Heading1("Tutorial")(), Paragraph("hello world")()]

Source code in ResearchToolkit/notion/block.py

def __init__(
    self, value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None
):
    """_summary_

    Args:
        value (_type_): _description_
        color (Optional[str], optional): _description_. Defaults to None.
        children (Optional[Union[dict, List[dict]]], optional): children should be output or list of `Block` object output.
                                                                e.g) Paragraph("hello world")()
                                                                    or [Heading1("Tutorial")(), Paragraph("hello world")()]
    """
    super(Paragraph, self).__init__(value, color)
    if children:
        if isinstance(children, dict):
            children = [children]
        self.out[self.name].update({"children": children})

NumberedListItem

NumberedListItem(value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None)

Bases: Paragraph

summary

Parameters:

• value (_type_) –

  description

• color (Optional[str], default: None ) –

  description. Defaults to None.

• children (Optional[Union[dict, List[dict]]], default: None ) –

  children should be output or list of Block object output. e.g) Paragraph("hello world")() or [Heading1("Tutorial")(), Paragraph("hello world")()]

Source code in ResearchToolkit/notion/block.py

def __init__(
    self, value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None
):
    """_summary_

    Args:
        value (_type_): _description_
        color (Optional[str], optional): _description_. Defaults to None.
        children (Optional[Union[dict, List[dict]]], optional): children should be output or list of `Block` object output.
                                                                e.g) Paragraph("hello world")()
                                                                    or [Heading1("Tutorial")(), Paragraph("hello world")()]
    """
    super(Paragraph, self).__init__(value, color)
    if children:
        if isinstance(children, dict):
            children = [children]
        self.out[self.name].update({"children": children})

ToDo

ToDo(value, checked: Optional[bool] = None, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None)

Bases: Paragraph

Source code in ResearchToolkit/notion/block.py

def __init__(
    self,
    value,
    checked: Optional[bool] = None,
    color: Optional[str] = None,
    children: Optional[Union[dict, List[dict]]] = None,
):
    super(ToDo, self).__init__(value, color, children)
    if checked:
        self.out[self.name].update({"checked": checked})

Toggle

Toggle(value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None)

Bases: Paragraph

summary

Parameters:

• value (_type_) –

  description

• color (Optional[str], default: None ) –

  description. Defaults to None.

• children (Optional[Union[dict, List[dict]]], default: None ) –

  children should be output or list of Block object output. e.g) Paragraph("hello world")() or [Heading1("Tutorial")(), Paragraph("hello world")()]

Source code in ResearchToolkit/notion/block.py

def __init__(
    self, value, color: Optional[str] = None, children: Optional[Union[dict, List[dict]]] = None
):
    """_summary_

    Args:
        value (_type_): _description_
        color (Optional[str], optional): _description_. Defaults to None.
        children (Optional[Union[dict, List[dict]]], optional): children should be output or list of `Block` object output.
                                                                e.g) Paragraph("hello world")()
                                                                    or [Heading1("Tutorial")(), Paragraph("hello world")()]
    """
    super(Paragraph, self).__init__(value, color)
    if children:
        if isinstance(children, dict):
            children = [children]
        self.out[self.name].update({"children": children})

Code

Code(value, caption: Optional[str] = None, language: Optional[str] = None)

Bases: Block

Source code in ResearchToolkit/notion/block.py

def __init__(self, value, caption: Optional[str] = None, language: Optional[str] = None):
    super(Code, self).__init__(value)
    self.out = {self.name: {"rich_text": [{"text": {"content": value}}]}}
    if caption:
        self.out[self.name].update({"caption": [{"text": {"content": caption}}]})
    if language.lower() in [
        "abap",
        "arduino",
        "bash",
        "basic",
        "c",
        "clojure",
        "coffeescript",
        "c++",
        "c#",
        "css",
        "dart",
        "diff",
        "docker",
        "elixir",
        "elm",
        "erlang",
        "flow",
        "fortran",
        "f#",
        "gherkin",
        "glsl",
        "go",
        "graphql",
        "groovy",
        "haskell",
        "html",
        "java",
        "javascript",
        "json",
        "julia",
        "kotlin",
        "latex",
        "less",
        "lisp",
        "livescript",
        "lua",
        "makefile",
        "markdown",
        "markup",
        "matlab",
        "mermaid",
        "nix",
        "objective-c",
        "ocaml",
        "pascal",
        "perl",
        "php",
        "plain text",
        "powershell",
        "prolog",
        "protobuf",
        "python",
        "r",
        "reason",
        "ruby",
        "rust",
        "sass",
        "scala",
        "scheme",
        "scss",
        "shell",
        "sql",
        "swift",
        "typescript",
        "vb.net",
        "verilog",
        "vhdl",
        "visual basic",
        "webassembly",
        "xml",
        "yaml",
        "java/c/c++/c#",
    ]:
        self.out[self.name].update({"language": language})

out instance-attribute

out = {name: {'rich_text': [{'text': {'content': value}}]}}

__bool__

__bool__()

Source code in ResearchToolkit/notion/block.py

def __bool__(self):
    return isinstance(self.value, str)

register

register(cls)

Source code in ResearchToolkit/notion/block.py

def register(cls):
    BLOCKS[cls.name] = cls
    return cls

make_rich_text_blocks

make_rich_text_blocks(name: str, text: str, **kwargs) -> List[Dict]

text를 block 길이 제한 만큼 잘라서 block들의 리스트로 변환합니다.

Parameters:

• name (str) –

  block 종류

• text (str) –

  컨텐츠 텍스트

Returns:

• List[Dict] –

  List[Dict]: append_block_children에 입력할 수 있는 block들

Source code in ResearchToolkit/notion/block.py

def make_rich_text_blocks(name: str, text: str, **kwargs) -> List[Dict]:
    """text를 block 길이 제한 만큼 잘라서 block들의 리스트로 변환합니다.

    Args:
        name (str): block 종류
        text (str): 컨텐츠 텍스트

    Returns:
        List[Dict]: append_block_children에 입력할 수 있는 block들
    """
    text_chuncks = [text[i : i + RICH_TEXT_LIMIT] for i in range(0, len(text), RICH_TEXT_LIMIT)]
    blocks = [BLOCKS[name](chunck, **kwargs)() for chunck in text_chuncks]
    return blocks

retrieve_block_children

retrieve_block_children(page_id: str, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None)

get all blocks on notion page. Args: page_id (str): notion page_id to search token_path (str, optional): description. Defaults to RESEARCH_API_PATH. token (Optional[str], optional): description. Defaults to None.

Returns:

• _type_ –

  request response code (200 if succeeded) and response text, block_children

Source code in ResearchToolkit/notion/block.py

def retrieve_block_children(
    page_id: str,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
):
    """
    get all blocks on notion page.
    Args:
        page_id (str): notion page_id to search
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.

    Returns:
        _type_: request response code (200 if succeeded) and response text, block_children
    """
    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
    }

    api_url = f"https://api.notion.com/v1/blocks/{page_id}/children?page_size=100"

    # get block children from notion
    response = requests.get(api_url, headers=headers)

    try:
        block_children = response.json()["results"]
    except AttributeError:
        print("failed to get block children")
        block_children = None

    return response.status_code, response.text, block_children

find_database_on_page

find_database_on_page(page_id: str) -> Dict[str, str]

find all database on page.

Returns:

• Dict[str, str] –

  Dict[str, str]: dictionary whose keys are database ids and values are database names

Source code in ResearchToolkit/notion/block.py

def find_database_on_page(page_id: str) -> Dict[str, str]:
    """
    find all database on page.

    Returns:
        Dict[str, str]: dictionary whose keys are database ids and values are database names

    """
    _, _, block_children = retrieve_block_children(page_id)

    databases = dict()
    if block_children is None:
        return databases
    for block in block_children:
        if block["type"] == "child_database":
            database = {block["id"]: block["child_database"]["title"]}
            databases.update(database)
    return databases

append_block_children

append_block_children(page_id: str, block: Union[dict, List[dict]], token_path: str = RESEARCH_API_PATH, token: Optional[str] = None)

append block or blocks to a page

Parameters:

• page_id (str) –

  description

• block (Union[dict, List[dict]]) –

  output or list of outputs of Block objects e.g) Paragraph("hello world")() or [Heading1("Tutorial")(), Paragraph("hello world")()]

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

Returns:

• _type_ –

  request response code (200 if succeeded) and response text

Source code in ResearchToolkit/notion/block.py

def append_block_children(
    page_id: str,
    block: Union[dict, List[dict]],
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
):
    """append block or blocks to a page

    Args:
        page_id (str): _description_
        block (Union[dict, List[dict]]): output or list of outputs of `Block` objects
                                         e.g) Paragraph("hello world")() or [Heading1("Tutorial")(), Paragraph("hello world")()]
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.

    Returns:
        _type_: request response code (200 if succeeded) and response text
    """
    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Notion-Version": "2022-06-28",
    }

    api_url = f"https://api.notion.com/v1/blocks/{page_id}/children"

    if isinstance(block, dict):
        block = [block]
    data = {"children": block}
    data = json.dumps(data)

    response = requests.request("PATCH", api_url, headers=headers, data=data)

    return response.status_code, response.text

common

RESEARCH_API_PATH module-attribute

RESEARCH_API_PATH = 'http://data.saige.in:8000/workspaces/notion/research_api_token.txt'

pascal_to_snake

pascal_to_snake(name)

Source code in ResearchToolkit/notion/common.py

def pascal_to_snake(name):
    return "_".join(re.sub(r"([A-Z])", r" \1", name).split()).lower()

get_token

get_token(token_path: str = RESEARCH_API_PATH)

Source code in ResearchToolkit/notion/common.py

def get_token(token_path: str = RESEARCH_API_PATH):
    if "http" in token_path:
        token = urlopen(token_path).read().decode("utf-8")
    else:
        with open(token_path, "r") as file:
            token = file.read()

    return token

use_saige_token_as_default

use_saige_token_as_default(function)

Source code in ResearchToolkit/notion/common.py

def use_saige_token_as_default(function):
    @wraps(function)
    def wrapper(
        token_path: str = RESEARCH_API_PATH,
        token: Optional[str] = None,
        *args,
        **kwargs,
    ):
        if not token:
            token = get_token(token_path)
        return function(token=token, *args, **kwargs)

    return wrapper

database

API for exchanging information with Notion database. Main features include, 1) creating a notion database with property assignments. 2) adding dictionary data to existing notion database.

FieldType module-attribute

FieldType = Dict[str, str]

RESPONSE_SUCCESS module-attribute

RESPONSE_SUCCESS = 200

QUERY_PAGE_SIZE module-attribute

QUERY_PAGE_SIZE = 100

convert_data_into_property

convert_data_into_property(data: Dict[str, Any], fields: FieldType)

converts dictionary data into a property form required by notion api. key of data and fields shoud match!

e.g) data = {"exp_name": "cls_base", "accuracy": 0.99} fields = {"exp_name": "title", "accuracy": "number"}

    return:
    {
        "exp_name": {
            "title": [
                "text": {"content": "cls_base"}
            ]
        },
        "accuracy": {
            "number": 0.99
        }
    }

Parameters:

• data (Dict[str, Any]) –

  key: property name, value: property value

• fields (FieldType) –

  key: property name, value: property type

Returns:

• _type_ –

  property required by notion api

Source code in ResearchToolkit/notion/database.py

def convert_data_into_property(data: Dict[str, Any], fields: FieldType):
    """converts dictionary data into a property form required by notion api.
       key of data and fields shoud match!

       e.g) data = {"exp_name": "cls_base", "accuracy": 0.99}
            fields = {"exp_name": "title", "accuracy": "number"}

            return:
            {
                "exp_name": {
                    "title": [
                        "text": {"content": "cls_base"}
                    ]
                },
                "accuracy": {
                    "number": 0.99
                }
            }

    Args:
        data (Dict[str, Any]): key: property name, value: property value
        fields (FieldType): key: property name, value: property type

    Returns:
        _type_: property required by notion api
    """
    properties = {}
    for key, value in data.items():
        if key not in fields:
            print(f"Invalid keys will be ignored: {key}")
            continue

        property_type = fields[key]

        if property_type not in PROPERTIES:
            print(f"property {property_type} not supported")
            continue

        if isinstance(value, dict):
            property_ = PROPERTIES[property_type](**value)
        else:
            property_ = PROPERTIES[property_type](value)

        if not property_:
            print(f"{value} is invalid for {property_type}")
            continue

        properties.update({key: property_()})

    return properties

make_database_property_schema

make_database_property_schema(fields: FieldType, relations: Dict[str, Dict[str, str]] = None) -> Dict

Parameters:

• fields (FieldType) –

  key: name of the property, value: property type (should be one of the keys in PROPERTIES) e.g) {"exp_name": "title", "accuracy": "number"} create a database whose property names are exp_name, accuracy and their property types are title and number.

• relations (Dict[str, Dict[str, str]], default: None ) –

  key: name of the relation property, value: dictionary with "database_id" and "type" info. "database_id" is notion database id to relate to "type" should be one of "single_property" or "dual_property". e.g) {"related_task": {"database_id": (database_id), "type": "dual_property}} !NOTE: relation database should also be shared with notion integration! Defaults to None.

Returns:

• Dict ( Dict ) –

  Notion API database property schema (https://developers.notion.com/reference/property-schema-object)

Source code in ResearchToolkit/notion/database.py

def make_database_property_schema(
    fields: FieldType,
    relations: Dict[str, Dict[str, str]] = None,
) -> Dict:
    """
    Args:
        fields (FieldType): key: name of the property, value: property type (should be one of the keys in `PROPERTIES`)
                                 e.g) {"exp_name": "title", "accuracy": "number"}
                                      create a database whose property names are exp_name, accuracy and their property types are title and number.
        relations (Dict[str, Dict[str, str]], optional): key: name of the relation property, value: dictionary with "database_id" and "type" info.
                                               "database_id" is notion database id to relate to
                                               "type" should be one of "single_property" or "dual_property".
                                               e.g) {"related_task": {"database_id": (database_id), "type": "dual_property}}
                                               !NOTE: relation database should also be shared with notion integration!
                                               Defaults to None.

    Returns:
        Dict: Notion API database property schema (https://developers.notion.com/reference/property-schema-object)
    """
    properties = {
        name: {
            property_.replace("text", "rich_text")
            .replace("person", "people")
            .replace("phone", "phone_number"): {}
        }
        for name, property_ in fields.items()
    }

    for name in properties.keys():
        for property_type in properties[name]:
            if property_type in ["select", "multi_select"]:
                properties[name][property_type] = {"options": []}

    if relations is not None:
        for name, info in relations.items():
            if info["type"] not in ["single_property", "dual_property"]:
                raise ValueError("relation type should be single_property or dual_property")
            properties.update(
                {
                    name: {
                        "relation": {
                            "database_id": info["database_id"],
                            info["type"]: {},
                        }
                    }
                }
            )

    return properties

_add_to_database

_add_to_database(database_id: str, properties: dict, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None)

add a page with properties to the notion database.

Parameters:

• database_id (str) –

  notion database id to add page.

• properties (dict) –

  page property dictionary in specific format described in notion api docs. see (https://developers.notion.com/reference/property-value-object). basic format: { "(property_name)":{ "(property_type)": (value whose type and structure are specific to property_type) } } This can be easily obtained by using child classes of BasicProperty. refer 'convert_data_into_property'.

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

Returns:

• _type_ –

  request response code (200 if succeeded) and response text, page_id (None if request failed)

Source code in ResearchToolkit/notion/database.py

def _add_to_database(
    database_id: str,
    properties: dict,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
):
    """add a page with properties to the notion database.

    Args:
        database_id (str): notion database id to add page.
        properties (dict): page property dictionary in specific format described in notion api docs.
                           see (https://developers.notion.com/reference/property-value-object).
                           basic format:
                           {
                                "(property_name)":{
                                     "(property_type)":
                                        (value whose type and structure are specific to property_type)
                                }
                            }
                           This can be easily obtained by using child classes of `BasicProperty`.
                           refer 'convert_data_into_property'.
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.

    Returns:
        _type_: request response code (200 if succeeded) and response text, page_id (None if request failed)
    """
    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Notion-Version": "2022-06-28",
    }

    api_url = "https://api.notion.com/v1/pages"

    data = {"parent": {"database_id": database_id}, "properties": properties}
    data = json.dumps(data)

    # create database in notion
    response = requests.request("POST", api_url, headers=headers, data=data)

    try:
        page_id = response.json()["id"]
    except AttributeError:
        print("database creation failed")
        page_id = None

    return response.status_code, response.text, page_id

create_database

create_database(page_id: str, fields: FieldType, relations: Dict[str, Dict[str, str]] = {}, title: Optional[str] = None, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None)

creates a notion database in a page_id page. it is recommended to add "user" field in fields which specifies the user of api when collaborating on a database.

Parameters:

• page_id (str) –

  notion page id

• fields (FieldType) –

  key: name of the property, value: property type (should be one of the keys in PROPERTIES) e.g) {"exp_name": "title", "accuracy": "number"} create a database whose property names are exp_name, accuracy and their property types are title and number.

• relations (Dict[str, Dict[str, str]], default: {} ) –

  key: name of the relation property, value: dictionary with "database_id" and "type" info. "database_id" is notion database id to relate to "type" should be one of "single_property" or "dual_property". e.g) {"related_task": {"database_id": (database_id), "type": "dual_property}} !NOTE: relation database should also be shared with notion integration!

• title (str, default: None ) –

  title of database. Defaults to None ("untitled" database will be created).

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

Returns:

• _type_ –

  request response code (200 if succeeded), response text, database_id (None if request failed)

Source code in ResearchToolkit/notion/database.py

def create_database(
    page_id: str,
    fields: FieldType,
    relations: Dict[str, Dict[str, str]] = {},
    title: Optional[str] = None,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
):
    """creates a notion database in a page_id page.
       it is recommended to add "user" field in `fields` which specifies the user of api when collaborating on a database.

    Args:
        page_id (str): notion page id
        fields (FieldType): key: name of the property, value: property type (should be one of the keys in `PROPERTIES`)
                                 e.g) {"exp_name": "title", "accuracy": "number"}
                                      create a database whose property names are exp_name, accuracy and their property types are title and number.
        relations (Dict[str, Dict[str, str]]): key: name of the relation property, value: dictionary with "database_id" and "type" info.
                                               "database_id" is notion database id to relate to
                                               "type" should be one of "single_property" or "dual_property".
                                               e.g) {"related_task": {"database_id": (database_id), "type": "dual_property}}
                                               !NOTE: relation database should also be shared with notion integration!
        title (str, optional): title of database. Defaults to None ("untitled" database will be created).
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.

    Returns:
        _type_: request response code (200 if succeeded), response text, database_id (None if request failed)
    """
    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Notion-Version": "2022-06-28",
    }

    api_url = "https://api.notion.com/v1/databases"

    # define a database
    property_types = fields.values()
    assert "title" in property_types, "title should be in properties"

    properties = make_database_property_schema(fields=fields, relations=relations)

    data = {"parent": {"type": "page_id", "page_id": page_id}, "properties": properties}

    if title:
        data.update(Title(title)())

    data = json.dumps(data)

    # create database in notion
    response = requests.request("POST", api_url, headers=headers, data=data)

    try:
        database_id = response.json()["id"]
    except AttributeError:
        print("database creation failed")
        database_id = None

    return response.status_code, response.text, database_id

retrieve_database_fields

retrieve_database_fields(database_id: str, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None, is_page: bool = False)

retrieves database fields

Parameters:

• database_id (_type_) –

  notion database id

• token_path (_type_, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (_type_, default: None ) –

  description. Defaults to None.

• is_page (bool, default: False ) –

  True if the id is for a page. Defaults to None.

Returns:

• _type_ –

  fields dictionary - key: property name, value: property type e.g) {"exp_name": "title", "accuracy": "number"}

Source code in ResearchToolkit/notion/database.py

def retrieve_database_fields(
    database_id: str,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
    is_page: bool = False,
):
    """retrieves database fields

    Args:
        database_id (_type_): notion database id
        token_path (_type_, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (_type_, optional): _description_. Defaults to None.
        is_page (bool): True if the id is for a page. Defaults to None.

    Returns:
        _type_: fields dictionary - key: property name, value: property type
                                      e.g) {"exp_name": "title", "accuracy": "number"}
    """
    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
    }

    sub_url = "pages" if is_page else "databases"
    api_url = f"https://api.notion.com/v1/{sub_url}/{database_id}"

    # get database from notion
    response = requests.get(api_url, headers=headers)

    # extract database fields
    database_fields = {
        name: prop["type"]
        .replace("rich_text", "text")
        .replace("people", "person")
        .replace("phone_number", "phone")
        for name, prop in response.json()["properties"].items()
    }

    return database_fields

add_data_to_database

add_data_to_database(database_id: str, data: dict, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None, user: Optional[Dict[str, BasicProperty]] = None)

add data to notion database. properties of data values are assigned to match the properties of existing database.

when collaborating on a database, it is recommended to use user option to specfiy who edited the database.

Parameters:

• database_id (str) –

  description

• data (dict) –

  data to store in database. keys should match names of the database properties

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

• user (Optional[Dict[str, BasicProperty]], default: None ) –

                          use when collaborating on a databaese to specify editor of the database.
                          key should be the property name of the database responsible for tracking api user.
                          if vaule is `Person`, a notification will be send in notion.
  

Returns:

• _type_ –

  request response code (200 if succeeded) and response text, page_id (None if request failed)

Source code in ResearchToolkit/notion/database.py

def add_data_to_database(
    database_id: str,
    data: dict,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
    user: Optional[Dict[str, BasicProperty]] = None,
):
    """
    add data to notion database.
    properties of data values are assigned to match the properties of existing database.

    when collaborating on a database, it is recommended to use `user` option to specfiy who edited the database.

    Args:
        database_id (str): _description_
        data (dict): data to store in database. keys should match names of the database properties
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.
        user (Optional[Dict[str, BasicProperty]], optional):
                                        use when collaborating on a databaese to specify editor of the database.
                                        key should be the property name of the database responsible for tracking api user.
                                        if vaule is `Person`, a notification will be send in notion.

    Returns:
        _type_: request response code (200 if succeeded) and response text, page_id (None if request failed)
    """
    database_fields = retrieve_database_fields(database_id, token_path=token_path, token=token)

    properties = convert_data_into_property(data, database_fields)

    if user and list(user.keys())[0] in database_fields:
        properties.update(user)

    return _add_to_database(database_id, properties, token_path=token_path, token=token)

query_database

query_database(database_id: str, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None, request_body: Optional[dict] = None)

query a database.
refer https://developers.notion.com/reference/post-database-query

Parameters:

• database_id (str) –

  description

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

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

  request body for query with filtering or sorting. Defaults to None. Check https://developers.notion.com/reference/post-database-query for detailed usage of BODY_PARAM.

Returns:

• _type_ –

  request response code (200 if succeeded), response text, results (None if request failed)

Source code in ResearchToolkit/notion/database.py

def query_database(
    database_id: str,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
    request_body: Optional[dict] = None,
):
    """
        query a database.
        refer https://developers.notion.com/reference/post-database-query

    Args:
        database_id (str): _description_
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.
        request_body (Optional[dict], optional): request body for query with filtering or sorting. Defaults to None.
            Check https://developers.notion.com/reference/post-database-query for detailed usage of BODY_PARAM.

    Returns:
        _type_: request response code (200 if succeeded), response text, results (None if request failed)
    """

    if not token:
        token = get_token(token_path)

    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
        "accept": "application/json",
        "content-type": "application/json",
    }

    api_url = f"https://api.notion.com/v1/databases/{database_id}/query"

    request_body = {
        "page_size": QUERY_PAGE_SIZE,
        **(request_body or {}),
    }

    results = None
    while True:
        response = requests.post(api_url, headers=headers, json=request_body)
        if response.status_code != RESPONSE_SUCCESS:
            raise RuntimeError(f"Notion Error: {response.text}")

        pages = response.json()["results"]
        if results is None:
            results = pages
        else:
            assert pages[0]["id"] == results[-1]["id"]
            results += pages[1:]

        if len(pages) < QUERY_PAGE_SIZE:
            break

        request_body["start_cursor"] = pages[-1]["id"]

    return response.status_code, response.text, results

get_property_value

get_property_value(property_)

Source code in ResearchToolkit/notion/database.py

def get_property_value(property_):
    property_type = property_["type"]
    property_type = (
        property_type.replace("rich_text", "text")
        .replace("people", "person")
        .replace("phone_number", "phone")
    )
    if property_type == "rollup":
        value = [get_property_value(p) for p in property_["rollup"]["array"]]
    else:
        if property_type in PROPERTIES:
            value = PROPERTIES[property_type].get_value(property_)
        else:
            # if the property type is not in PROPERTIES, results can be less specific.
            value = BasicProperty.get_value(property_)
    return value

get_database_in_pd_dataframe

get_database_in_pd_dataframe(database_id: str, token_path: str = RESEARCH_API_PATH, token: Optional[str] = None, request_body: Optional[dict] = None)

get notion database and return it in pandas dataframe.

Parameters:

• database_id (str) –

  description

• token_path (str, default: RESEARCH_API_PATH ) –

  description. Defaults to RESEARCH_API_PATH.

• token (Optional[str], default: None ) –

  description. Defaults to None.

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

  request body for query with filtering or sorting. Defaults to None. Check https://developers.notion.com/reference/post-database-query for detailed usage of BODY_PARAM.

Returns:

• –

  pd.DataFrame: description

Source code in ResearchToolkit/notion/database.py

def get_database_in_pd_dataframe(
    database_id: str,
    token_path: str = RESEARCH_API_PATH,
    token: Optional[str] = None,
    request_body: Optional[dict] = None,
):
    """
    get notion database and return it in pandas dataframe.

    Args:
        database_id (str): _description_
        token_path (str, optional): _description_. Defaults to RESEARCH_API_PATH.
        token (Optional[str], optional): _description_. Defaults to None.
        request_body (Optional[dict], optional): request body for query with filtering or sorting. Defaults to None.
            Check https://developers.notion.com/reference/post-database-query for detailed usage of BODY_PARAM.

    Returns:
        pd.DataFrame: _description_
    """
    status, text, results = query_database(database_id, token_path, token, request_body)

    data = {key: [] for key in results[0]["properties"].keys()}
    for result in results:
        properties = result["properties"]
        for key, property_ in properties.items():
            value = get_property_value(property_)
            data[key].append(value)

    return pd.DataFrame(data)

update_page_data

update_page_data(token: str, page_id: str, data: Dict[str, Any])

Database page의 data를 업데이트 합니다.

Source code in ResearchToolkit/notion/database.py

@use_saige_token_as_default
def update_page_data(
    token: str,
    page_id: str,
    data: Dict[str, Any],
):
    """Database page의 data를 업데이트 합니다."""
    database_fields = retrieve_database_fields(database_id=page_id, token=token, is_page=True)
    properties = convert_data_into_property(data=data, fields=database_fields)

    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
        "accept": "application/json",
        "content-type": "application/json",
    }

    api_url = f"https://api.notion.com/v1/pages/{page_id}"
    data = {"properties": properties}

    response = requests.patch(api_url, headers=headers, json=data)
    return response.status_code, response.text

update_database_fields

update_database_fields(token: str, database_id: str, fields: FieldType)

Database fields (properties) 를 업데이트 합니다.

Source code in ResearchToolkit/notion/database.py

@use_saige_token_as_default
def update_database_fields(
    token: str,
    database_id: str,
    fields: FieldType,
):
    """Database fields (properties) 를 업데이트 합니다."""
    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
        "accept": "application/json",
        "content-type": "application/json",
    }

    api_url = f"https://api.notion.com/v1/databases/{database_id}"

    properties = make_database_property_schema(fields=fields)
    data = {"properties": properties}
    response = requests.patch(api_url, headers=headers, json=data)

    return response.status_code, response.text

update_database_fields_safe

update_database_fields_safe(token: str, database_id: str, fields: FieldType)

Database fields (properties) 를 업데이트 합니다. 데이터베이스에 동일한 이름으로 다른 타입의 프로퍼티가 존재하면 에러를 레이즈 합니다.

Source code in ResearchToolkit/notion/database.py

@use_saige_token_as_default
def update_database_fields_safe(
    token: str,
    database_id: str,
    fields: FieldType,
):
    """Database fields (properties) 를 업데이트 합니다. 데이터베이스에 동일한 이름으로 다른 타입의 프로퍼티가 존재하면 에러를 레이즈 합니다."""
    original_fields = retrieve_database_fields(database_id)
    fields_to_update = {}
    for key, value in fields.items():
        if key not in original_fields:
            fields_to_update.update({key: value})
        elif value != original_fields[key]:
            raise ValueError(
                f"Notion database field `{key}` already exists with type `{original_fields[key]}`."
            )
    if fields_to_update:
        status, message = update_database_fields(
            token=token,
            database_id=database_id,
            fields=fields_to_update,
        )
        if status != RESPONSE_SUCCESS:
            raise RuntimeError(f"Notion Error: {message}")

filter

Filter

Filter(filter_: dict)

Filter class for handling Notion Filter object. Please see https://developers.notion.com/reference/post-database-query-filter for detail.

You can now query to notion DB with filtering as following:

import SaigeToolkit.SaigeToolkit.notion as notion
filter_1 = notion.property.Select.get_filter("select_col", "equals", "sel1")
filter_2 = notion.property.Number.get_filter("number_col", "does_not_equal", 10)
filter_3 = notion.property.Checkbox.get_filter("checkbox_col", "does_not_equal", True)
filter_4 = notion.filter.Filter.compound("and", [filter_1, filter_2, filter_3])
_, _, results = notion.database.query_database(
    db_id,
    RESEARCH_API_PATH,
    request_body=filter_4.to_dict()
)

See notion.so/8cd58f07a254405ebb1bc327d690c8aa for more examples.

Source code in ResearchToolkit/notion/filter.py

def __init__(self, filter_: dict):
    assert isinstance(filter_, dict)
    self.filter = filter_

filter instance-attribute

filter = filter_

to_dict

to_dict()

Source code in ResearchToolkit/notion/filter.py

def to_dict(self):
    keys = list(self.filter.keys())
    if "property" in keys:
        return {"filter": self.filter}
    assert len(keys) == 1 and keys[0] in ["or", "and"]
    return {
        "filter": {keys[0]: [filter_.to_dict().get("filter") for filter_ in self.filter[keys[0]]]}
    }

compound classmethod

compound(op: str, filters: List[object])

Source code in ResearchToolkit/notion/filter.py

@classmethod
def compound(cls, op: str, filters: List[object]):
    assert op in {"or", "and"}
    return cls({op: filters})

get_avaliable_filter_ops

get_avaliable_filter_ops(property_name: str)

Source code in ResearchToolkit/notion/filter.py

def get_avaliable_filter_ops(property_name: str):

    none_type = type(None)

    text_filter_ops = {
        "equals": str,
        "does_not_equal": str,
        "contains": str,
        "does_not_contain": str,
        "starts_with": str,
        "ends_with": str,
        "is_empty": bool,
        "is_not_empty": bool,
    }

    number_filter_ops = {
        "equals": (int, float),
        "does_not_equal": (int, float),
        "greater_than": (int, float),
        "less_than": (int, float),
        "greater_than_or_equal_to": (int, float),
        "less_than_or_equal_to": (int, float),
        "is_empty": bool,
        "is_not_empty": bool,
    }

    checkbox_filter_ops = {"equals": bool, "does_not_equal": bool}

    select_filter_ops = {
        "equals": str,
        "does_not_equal": str,
        "is_empty": bool,
        "is_not_empty": bool,
    }

    multi_select_filter_ops = {
        "contains": str,
        "does_not_contain": str,
        "is_empty": bool,
        "is_not_empty": bool,
    }

    date_filter_ops = {
        "equals": str,
        "before": str,
        "after": str,
        "on_or_before": str,
        "is_empty": bool,
        "is_not_empty": bool,
        "on_or_after": str,
        "past_week": none_type,
        "past_month": none_type,
        "this_week": none_type,
        "next_week": none_type,
        "next_month": none_type,
        "next_year": none_type,
    }

    files_filter_ops = {
        "is_empty": bool,
        "is_not_empty": bool,
    }

    return {
        "title": text_filter_ops,
        "rich_text": text_filter_ops,
        "url": text_filter_ops,
        "email": text_filter_ops,
        "phone_number": text_filter_ops,
        "number": number_filter_ops,
        "checkbox": checkbox_filter_ops,
        "select": select_filter_ops,
        "multi_select": multi_select_filter_ops,
        "status": select_filter_ops,
        "date": date_filter_ops,
        "people": multi_select_filter_ops,
        "files": files_filter_ops,
        "relation": multi_select_filter_ops,
    }[property_name]

property

PROPERTIES module-attribute

PROPERTIES = {}

BasicProperty

BasicProperty(value, **kwargs)

Notion database basic properties are implemented. It converts values into property object structures specified by the notion api. Additionally, it can check whether given value is valid for the property. Note that status property is not implemented due to the limited notion api capabilites.

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

value instance-attribute

value = value

__init_subclass__

__init_subclass__(**kwargs)

Source code in ResearchToolkit/notion/property.py

def __init_subclass__(cls, **kwargs):
    super().__init_subclass__(**kwargs)
    name = pascal_to_snake(cls.__name__)
    cls.name = (
        name.replace("text", "rich_text")
        .replace("person", "people")
        .replace("phone", "phone_number")
    )
    cls.filter_ops = get_avaliable_filter_ops(cls.name)
    register(cls, name)

__call__

__call__()

Source code in ResearchToolkit/notion/property.py

def __call__(self):
    return {self.name: self.value}

get_value classmethod

get_value(property_: dict)

extracts values from a property object structure specified by the notion api. (opposite operation of call method)

Parameters:

• property_ (dict) –

  property dictionary output of notion api. must include "type" key.

Returns:

• _type_ –

  value of given property. ex) 3.14 in above example.

Source code in ResearchToolkit/notion/property.py

@classmethod
def get_value(cls, property_: dict):
    """
    extracts values from a property object structure specified by the notion api.
    (opposite operation of __call__ method)


    Args:
        property_ (dict): property dictionary output of notion api. must include "type" key.
        ex)
        {
            "type": "number",
            "number": 3.14
        }

    Returns:
        _type_: value of given property. ex) 3.14 in above example.
    """
    return property_[property_["type"]]

get_filter classmethod

get_filter(property_name, filter_op, filter_val)

Source code in ResearchToolkit/notion/property.py

@classmethod
def get_filter(cls, property_name, filter_op, filter_val):
    assert filter_op in cls.filter_ops.keys()
    assert isinstance(filter_val, cls.filter_ops[filter_op])
    return Filter({"property": property_name, f"{cls.name}": {filter_op: filter_val}})

Text

Text(value, **kwargs)

Bases: BasicProperty

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    return isinstance(self.value, str)

__call__

__call__()

Source code in ResearchToolkit/notion/property.py

def __call__(self):
    return {self.name: [{"text": {"content": self.value}}]}

get_value classmethod

get_value(property_: dict)

Source code in ResearchToolkit/notion/property.py

@classmethod
def get_value(cls, property_: dict):
    return "".join([text[text["type"]]["content"] for text in property_[property_["type"]]])

Title

Title(value, **kwargs)

Bases: Text

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

Number

Number(value, **kwargs)

Bases: BasicProperty

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    return isinstance(self.value, (float, int))

Select

Select(value, **kwargs)

Bases: BasicProperty

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    return isinstance(self.value, str)

__call__

__call__()

Source code in ResearchToolkit/notion/property.py

def __call__(self):
    return {self.name: {"name": self.value}}

get_value classmethod

get_value(property_: dict)

Source code in ResearchToolkit/notion/property.py

@classmethod
def get_value(cls, property_: dict):
    if property_["select"] is None:
        return None
    else:
        return property_["select"]["name"]

MultiSelect

MultiSelect(value, **kwargs)

Bases: BasicProperty

value should be a string or list of strings.

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    return isinstance(self.value, list) or isinstance(self.value, str)

__call__

__call__()

Source code in ResearchToolkit/notion/property.py

def __call__(self):
    if not isinstance(self.value, list):
        value = [self.value]
    else:
        value = self.value
    return {self.name: [{"name": v} for v in value]}

get_value classmethod

get_value(property_: dict)

Source code in ResearchToolkit/notion/property.py

@classmethod
def get_value(cls, property_: dict):
    return [p["name"] for p in property_["multi_select"]]

Date

Date(value, **kwargs)

Bases: BasicProperty

date should be in iso-format.

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    def check_isoformat(date):
        try:
            datetime.fromisoformat(date)
            return True
        except ValueError:
            print("date is not in iso-format")
            return False

    if isinstance(self.value, (tuple, list)):
        if len(self.value) > 2:
            return False
        else:
            return all(check_isoformat(date) for date in self.value)
    else:
        return check_isoformat(self.value)

__call__

__call__()

Source code in ResearchToolkit/notion/property.py

def __call__(self):
    if isinstance(self.value, (tuple, list)):
        if len(self.value) == 2:
            date = {"start": self.value[0], "end": self.value[1]}
        else:
            date = {"start": self.value[0]}
    else:
        date = {"start": self.value}
    return {self.name: date}

Person

Person(value, **kwargs)

Bases: BasicProperty

value should be either name or email used in Notion.

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

users class-attribute instance-attribute

users = get_notion_users()

user_names class-attribute instance-attribute

user_names = {name: (info['id']) for name, info in (items())}

user_emails class-attribute instance-attribute

user_emails = {(info['email']): (info['id']) for info in (values())}

validate_member

validate_member(person)

Source code in ResearchToolkit/notion/property.py

def validate_member(self, person):
    if person in self.user_names or person in self.user_emails:
        return True
    else:
        return False

get_id

get_id(person)

Source code in ResearchToolkit/notion/property.py

def get_id(self, person):
    if "@" in person:
        id_ = self.user_emails[person]
    else:
        id_ = self.user_names[person]
    return id_

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    if isinstance(self.value, list):
        return all(self.validate_member(v) for v in self.value)
    else:
        return self.validate_member(self.value)

__call__

__call__()

Source code in ResearchToolkit/notion/property.py

def __call__(self):
    if not isinstance(self.value, list):
        value = [self.value]
    else:
        value = self.value
    return {self.name: [{"id": self.get_id(v)} for v in value]}

get_person_with_user_info classmethod

get_person_with_user_info(user_name=None, user_email=None)

Source code in ResearchToolkit/notion/property.py

@classmethod
def get_person_with_user_info(cls, user_name=None, user_email=None):
    for user_info in [user_name, user_email]:
        person = cls(user_info)
        if person:
            return person

    raise Exception("user cannot be identifed")

get_value classmethod

get_value(property_: dict)

Source code in ResearchToolkit/notion/property.py

@classmethod
def get_value(cls, property_: dict):
    return [p["name"] for p in property_["people"]]

Files

Files(value, file_name='unknown')

Bases: BasicProperty

only files on remotes can be uploaded.

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, file_name="unknown"):
    super(Files, self).__init__(value)
    self.file_name = file_name

file_name instance-attribute

file_name = file_name

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    response = requests.get(self.value)
    return response.status_code == 200

__call__

__call__()

Source code in ResearchToolkit/notion/property.py

def __call__(self):
    return {self.name: [{"name": self.file_name, "file": {"url": self.value}}]}

get_value classmethod

get_value(property_: dict)

Source code in ResearchToolkit/notion/property.py

@classmethod
def get_value(cls, property_: dict):
    return [p["file"]["url"] for p in property_["files"]]

Checkbox

Checkbox(value, **kwargs)

Bases: BasicProperty

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    return isinstance(self.value, bool)

Url

Url(value, **kwargs)

Bases: BasicProperty

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    return isinstance(self.value, str)

Email

Email(value, **kwargs)

Bases: BasicProperty

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    return isinstance(self.value, str)

Phone

Phone(value, **kwargs)

Bases: BasicProperty

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    return isinstance(self.value, str)

Relation

Relation(value, **kwargs)

Bases: BasicProperty

value should be a notion page id

Source code in ResearchToolkit/notion/property.py

def __init__(self, value, **kwargs):
    self.value = value

__bool__

__bool__()

Source code in ResearchToolkit/notion/property.py

def __bool__(self):
    token = get_token(RESEARCH_API_PATH)

    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
    }

    api_url = f"https://api.notion.com/v1/pages/{self.value}"

    response = requests.get(api_url, headers=headers)

    return response.status_code == 200

__call__

__call__()

Source code in ResearchToolkit/notion/property.py

def __call__(self):
    return {self.name: [{"id": self.value}]}

get_value classmethod

get_value(property_: dict)

Source code in ResearchToolkit/notion/property.py

@classmethod
def get_value(cls, property_: dict):
    return [p["id"] for p in property_["relation"]]

register

register(cls, name)

Source code in ResearchToolkit/notion/property.py

def register(cls, name):
    PROPERTIES[name] = cls
    return cls

user

MEMBER_INFO_PATH module-attribute

MEMBER_INFO_PATH = '/NFS/workspaces/notion/git_to_notion_user.json'

get_notion_users

get_notion_users()

Source code in ResearchToolkit/notion/user.py

def get_notion_users():
    token = get_token(RESEARCH_API_PATH)

    headers = {
        "Authorization": f"Bearer {token}",
        "Notion-Version": "2022-06-28",
    }

    api_url = "https://api.notion.com/v1/users"

    response = requests.get(api_url, headers=headers)
    result = response.json()["results"]

    users = {
        user["name"]: {"id": user["id"], "email": user["person"]["email"]}
        for user in result
        if "person" in user
    }

    return users

get_git_user

get_git_user(repo_path)

get git user information (name and email) useful when collaborators share a database. can automatically specfiy editor of the database using git user info

Parameters:

• repo_path (_type_) –

  git repo path

Returns:

• _type_ –

  description

Source code in ResearchToolkit/notion/user.py

def get_git_user(repo_path):
    """get git user information (name and email)
       useful when collaborators share a database.
       can automatically specfiy editor of the database using git user info

    Args:
        repo_path (_type_): git repo path

    Returns:
        _type_: _description_
    """
    r = git.Repo.init(repo_path)
    reader = r.config_reader()
    user_name = reader.get_value("user", "name")
    user_email = reader.get_value("user", "email")
    return user_name, user_email

find_git

find_git(path)

find directory with .git file in parent directories. return None if not found.

Parameters:

• path (_type_) –

  description

Returns:

• _type_ –

  description

Source code in ResearchToolkit/notion/user.py

def find_git(path):
    """find directory with .git file in parent directories.
       return None if not found.

    Args:
        path (_type_): _description_

    Returns:
        _type_: _description_
    """
    if os.path.isdir(os.path.join(path, ".git")):
        return path
    else:
        if path in ["", "/"]:
            return None
        else:
            return find_git(os.path.dirname(path))