Notion APIで大量のページをリレーション付きで一括投稿する

# Ryeの場合
$ rye add notion-client

# pipの場合
$ pip install notion-client

import os
from notion_client import Client

client = Client(auth=os.environ["NOTION_TOKEN"])

def get_db_names_and_ids_from_query(client, query: str) -> list:
    """
    Search for databases in Notion using a query string and 
    return a list of dictionaries containing database names and IDs.

    Args:
        client: The Notion client object.
        query (str): The search query string.

    Returns:
        list: A list of dictionaries, each containing 'db_title' and 'db_id' for found databases.
        If no databases are found, returns an empty list.
    """
    results = client.search(query=query).get("results")

    # Dict in List for db_info. [{db_title: db_id}, ...]
    db_names_ids = []
    db_count = 0

    for result in results:
        # Check if the result is a database
        if result["object"] == "database":
            db_count += 1
            db_info = {}
            db_info["db_title"] = result["title"][0]["plain_text"]
            db_info["db_id"] = result["id"]
            db_names_ids.append(db_info)
    
    if db_count:
        print(f"Found {db_count} databases.")
        return db_names_ids
    else:
        print(f"No databases found with query: {query}")
        return []


# Retrieve Parent DB
parent_db_name = "元DB"
parent_db_results = get_db_names_and_ids_from_query(client, parent_db_name)
PARENT_DB_ID = parent_db_results[0]["db_id"]

# Retrieve child DBs
child_db_results = get_db_names_and_ids_from_query(client, "リレーション先")

# Store
CHILD_DB_A_ID = db_results[0]["db_id"]
CHILD_DG_B_ID = db_results[1]["db_id"]

def get_all_pages_info_from_db(client, database_id: str) -> list:
    """
    Retrieves all information from a specified Notion database.

    Args:
        client: The Notion client object.
        database_id (str): The ID of the Notion database to query.

    Returns:
        list: A list of dictionaries, each containing all information for a page in the database.
    """
    response = client.databases.query(
        **{
            "database_id": database_id,
        }
    )

    return response["results"]


# 子DB_AのIDをもとに、DB内のページ情報をすべて取得
# Get all pages info from child_db_A
pages_info_A = get_all_pages_info_from_db(client, CHILD_DB_A_ID)

# 取得したページ情報一覧から、ページのIDのみをそれぞれ抜き出して辞書inリストの形で保持
# child_db_A_pages: {item_1: page_id, item_2: page_id, ...}
child_db_A_pages = {
    page["properties"]["Name"]["title"][0]["text"]["content"]: page["id"]
    for page in pages_info_A
}

# 子DB_Bについても繰り返す
# Get all pages info from child_db_B
pages_info_B = get_all_pages_info_from_db(client, CHILD_DB_B_ID)

# child_db_B_pages: {item_A: page_id, item_B: page_id, ...}
child_db_B_pages = {
    page["properties"]["Name"]["title"][0]["text"]["content"]: page["id"]
    for page in pages_info_B
}

### Multi Select Property

# Define the single tag property by a each color
gray = {
    'color': 'gray',
    'description': None,
    'id': 'gray_id',
    'name': 'gray'
}

orange = {
    'color': 'orange',
    'description': None,
    'id': 'orange_id',
    'name': 'orange'
}

yellow = {
    'color': 'yellow',
    'description': None,
    'id': 'yellow_id',
    'name': 'yellow'
}

red = {
    'color': 'red',
    'description': None,
    'id': 'red_id',
    'name': 'red'
}

green = {
    'color': 'green',
    'description': None,
    'id': 'green_id',
    'name': 'green'
}

blue = {
    'color': 'blue',
    'description': None,
    'id': 'blue_id',
    'name': 'blue'
}

# Get the defined colors properties based on the provided color names.
def get_multi_select_properties_by_color_names(*color_names) -> list:
    """
    Get the defined colors based on the provided color names.
    If no color names are provided, return all defined colors.
    """
    color_dict = {
        'gray': gray,
        'orange': orange,
        'yellow': yellow,
        'red': red,
        'green': green,
        'blue': blue
    }

    # If no color names are provided, return all colors
    if not color_names:
        return list(color_dict.values())
    
    # Return the colors that match the provided color names
    result = []
    for name in color_names:
        if name in color_dict:
            result.append(color_dict[name])
        # If the color name is not found, print a warning message
        else:
            print(f"Warning: '{name}' is not a valid color name and will be ignored.")
    
    return result

## Define the Multi Select property value for the parent database
# Prepare some patterns
multi_select_properties_all = get_multi_select_properties_by_color_names()

multi_select_properties_pattern_1 = get_multi_select_properties_by_color_names('red', 'blue', 'green')

multi_select_properties_pattern_2 = get_multi_select_properties_by_color_names('blue', 'yellow', 'orange', 'gray')

multi_select_properties_pattern_3 = get_multi_select_properties_by_color_names('red', 'green', 'yellow')

def create_page_property_2post_parent_db(post_title: str, post_text: str, multi_select_properties: list, child_page_A_ids: list, child_page_B_ids: list) -> dict:
    """
    Create a page property for the parent database.
    """
    page_property = {
        'Name': {
            'id': 'title',
            'type': 'title',
            "title": [{ "type": "text", "text": { "content": post_title } }]
        },
        'テキストフィールド': {
            'id': 'text_field_id',
            'type': 'rich_text',
            'rich_text': [{
                "type": "text",
                "text": {
                    "content": post_text,
                    "link": None
                }
            }],
        },
        'テスト複数セレクト': {
           'multi_select': multi_select_properties
        },
        'リレーション先DB_A': {
            # 元DBのリレーション列ID
            'id': 'relation_column_a_id',
            # リレーション先DBのページIDをリスト形式で追加
            'relation': [{"id": child_id} for child_id in child_page_A_ids],
            'type': 'relation'
        },
        'リレーション先DB_B': {
            'id': 'relation_column_b_id',
            'relation': [{"id": child_id} for child_id in child_page_B_ids],
            'type': 'relation'
        }
    }
    
    return page_property

# Example usage:
# page_prop = create_page_property("Post Title", "Post Text", [{"name": "Tag1"}, {"name": "Tag2"}], ["id1", "id2"], ["id3", "id4"])

## Define other properties for the parent database
# Create multiple page properties for the parent database
num_pages = 20000  # Number of pages to create

for i in range(num_pages):
    # Randomly select multi_select_properties
    multi_select_properties = random.choice([
        multi_select_properties_all,
        multi_select_properties_pattern_1,
        multi_select_properties_pattern_2,
        multi_select_properties_pattern_3
    ])

    # Randomly select child_page_A_ids (1 to 5 items)
    child_A_ids = random.sample(list(child_db_A_pages.values()), random.randint(1, 5))

    # Randomly select child_page_B_ids (1 to 5 items)
    child_B_ids = random.sample(list(child_db_B_pages.values()), random.randint(1, 5))

    page_property = create_page_property_2post_parent_db(
        post_title=f"Performance Test {i+1}",
        post_text=f"This is performance test {i+1} for the bulk upload.",
        multi_select_properties=multi_select_properties,
        child_page_A_ids=child_A_ids,
        child_page_B_ids=child_B_ids
    )

    # Post a single page to the parent database with retry logic
    for attempt in range(5):
        try:
            response = client.pages.create(
                parent={"database_id": PARENT_DB_ID},
                properties=page_property
            )
            break  # If successful, break out of the retry loop
        except Exception as e:
            if attempt < 4:  # If it's not the last attempt
                print(f"Error occurred: {e}. Retrying in 60 seconds...")
                time.sleep(60)
            else:
                error_message = f"Failed to create page after 5 attempts. Error: {e}"
                raise RequestException(error_message) from e

    # Print the result
    print(f"Successfully created page {i+1}")

    # Insert a pause after each page creation
    time.sleep(3)

# Output the result
print(f"Successfully created {num_pages} pages!")

import os
from pprint import pprint
import random
import time

from notion_client import Client

# Import my functions
import get_all_pages_info_from_db
import get_multi_select_properties_by_color_names
import create_page_property_2post_parent_db

client = Client(auth=your_notion_token)

# Database IDs for testing
PARENT_DB_ID = 'your_parent_db_id'
CHILD_DB_A_ID = 'your_child_db_a_id'
CHILD_DB_B_ID = 'your_child_db_b_id'

# Get all pages info from child_db_A
pages_info_A = get_all_pages_info_from_db(client, CHILD_DB_A_ID)

# child_db_A_pages: {item_1: page_id, item_2: page_id, ...}
child_db_A_pages = {
    page["properties"]["Name"]["title"][0]["text"]["content"]: page["id"]
    for page in pages_info_A
}

# Get all pages info from child_db_B
pages_info_B = get_all_pages_info_from_db(client, CHILD_DB_B_ID)

# child_db_B_pages: {item_A: page_id, item_B: page_id, ...}
child_db_B_pages = {
    page["properties"]["Name"]["title"][0]["text"]["content"]: page["id"]
    for page in pages_info_B
}

## Define the Multi Select property value for the parent database
# Prepare some patterns
multi_select_properties_all = get_multi_select_properties_by_color_names()

multi_select_properties_pattern_1 = get_multi_select_properties_by_color_names('red', 'blue', 'green')

multi_select_properties_pattern_2 = get_multi_select_properties_by_color_names('blue', 'yellow', 'orange', 'gray')

multi_select_properties_pattern_3 = get_multi_select_properties_by_color_names('red', 'green', 'yellow')


## Define other properties for the parent database
# Create multiple page properties for the parent database
num_pages = 20000  # Number of pages to create

for i in range(num_pages):
    # Randomly select multi_select_properties
    multi_select_properties = random.choice([
        multi_select_properties_all,
        multi_select_properties_pattern_1,
        multi_select_properties_pattern_2,
        multi_select_properties_pattern_3
    ])

    # Randomly select child_page_A_ids (1 to 5 items)
    child_A_ids = random.sample(list(child_db_A_pages.values()), random.randint(1, 5))

    # Randomly select child_page_B_ids (1 to 5 items)
    child_B_ids = random.sample(list(child_db_B_pages.values()), random.randint(1, 5))

    page_property = create_page_property_2post_parent_db(
        post_title=f"Performance Test {i+1}",
        post_text=f"This is performance test {i+1} for the bulk upload.",
        multi_select_properties=multi_select_properties,
        child_page_A_ids=child_A_ids,
        child_page_B_ids=child_B_ids
    )

    # Post a single page to the parent database with retry logic
    for attempt in range(5):
        try:
            response = client.pages.create(
                parent={"database_id": PARENT_DB_ID},
                properties=page_property
            )
            break  # If successful, break out of the retry loop
        except Exception as e:
            if attempt < 4:  # If it's not the last attempt
                print(f"Error occurred: {e}. Retrying in 60 seconds...")
                time.sleep(60)
            else:
                error_message = f"Failed to create page after 5 attempts. Error: {e}"
                raise RequestException(error_message) from e

    # Print the result
    print(f"Successfully created page {i+1}")

    # Insert a pause after each page creation
    time.sleep(3)

# Output the result
print(f"Successfully created {num_pages} pages!")

def get_all_pages_info_from_db(client, database_id: str) -> list:
    """
    Retrieves all information from a specified Notion database.

    Args:
        client: The Notion client object.
        database_id (str): The ID of the Notion database to query.

    Returns:
        list: A list of dictionaries, each containing all information for a page in the database.
    """
    response = client.databases.query(
        **{
            "database_id": database_id,
        }
    )

    return response["results"]


# Retrieve a list of page IDs of records existing in the database
def get_page_ids_from_db(client, database_id):
    """
    Retrieves a list of page IDs from a specified Notion database.

    Args:
        client: The Notion client object.
        database_id: The ID of the Notion database to query.

    Returns:
        list: A list of page IDs from the database.
    """
    response = client.databases.query(
        **{
            "database_id": database_id,
        }
    )

    # Extract only results from the whole response
    results = response["results"]
    page_ids = []
    for result in results:
        # Append each page ID to the list
        page_ids.append(result["id"])

    print(f"read_pages_from_database completed. (len={len(page_ids)})")
    return page_ids


def get_page_ids_and_titles_from_db(client, database_id: str) -> list: 
    """
    Retrieves page IDs and titles from a specified Notion database.

    Args:
        client: The Notion client object.
        database_id (str): The ID of the Notion database to query.

    Returns:
        list: A list of dictionaries, each containing 'page_id' and 'page_title' for each page in the database.
              If a page has no title, 'page_title' will be None.
    """
    response = client.databases.query(
        **{
            "database_id": database_id,
        }
    )

    # Extract only results from the whole response
    results = response["results"]
    page_datum = []
    for result in results:
        page_id = result["id"]
        try:
            page_title = result["properties"]["Name"]["title"][0]["plain_text"]
        except IndexError:
            page_title = None
        page_datum.append({"page_id": page_id, "page_title": page_title})

    return page_datum


def get_db_names_and_ids_from_query(client, query: str) -> list:
    """
    Search for databases in Notion using a query string and return a list of dictionaries containing database names and IDs.

    Args:
        client: The Notion client object.
        query (str): The search query string.

    Returns:
        list: A list of dictionaries, each containing 'db_title' and 'db_id' for found databases.
        If no databases are found, returns an empty list.
    """
    results = client.search(query=query).get("results")

    # Dict in List for db_info. [{db_title: db_id}, ...]
    db_names_ids = []
    db_count = 0

    for result in results:
        # Check if the result is a database
        if result["object"] == "database":
            db_count += 1
            db_info = {}
            db_info["db_title"] = result["title"][0]["plain_text"]
            db_info["db_id"] = result["id"]
            db_names_ids.append(db_info)
    
    if db_count:
        print(f"Found {db_count} databases.")
        return db_names_ids
    else:
        print(f"No databases found with query: {query}")
        return []


def get_db_properties_from_db_title_and_id(client, db_title: str, db_id: str) -> dict:
    """
    Retrieve the properties of a Notion database using its title and ID.
    CAUTION: The related database IDs that is retrieved are the 'Database ID's, not the 'Page ID's.

    Args:
        client: The Notion client object.
        db_title (str): The title of the database to search for.
        db_id (str): The ID of the database to match.

    Returns:
        dict: A dictionary containing the properties of the found database.
        If no matching database is found, returns an empty dictionary.
    """
    results = client.search(query=db_title).get("results")

    for result in results:
        if result["object"] == "database" and result['id'] == db_id:
            return result["properties"]
    
    print(f"No database found with title: {db_title} and id: {db_id}")
    return {}

### Multi Select Property

# Define the single tag property by a each color
gray = {
    'color': 'gray',
    'description': None,
    'id': 'gray_id',
    'name': 'gray'
}

orange = {
    'color': 'orange',
    'description': None,
    'id': 'orange_id',
    'name': 'orange'
}

yellow = {
    'color': 'yellow',
    'description': None,
    'id': 'yellow_id',
    'name': 'yellow'
}

red = {
    'color': 'red',
    'description': None,
    'id': 'red_id',
    'name': 'red'
}

green = {
    'color': 'green',
    'description': None,
    'id': 'green_id',
    'name': 'green'
}

blue = {
    'color': 'blue',
    'description': None,
    'id': 'blue_id',
    'name': 'blue'
}


# Get the defined colors properties based on the provided color names.
def get_multi_select_properties_by_color_names(*color_names) -> list:
    """
    Get the defined colors based on the provided color names.
    If no color names are provided, return all defined colors.
    """
    color_dict = {
        'gray': gray,
        'orange': orange,
        'yellow': yellow,
        'red': red,
        'green': green,
        'blue': blue
    }

    # If no color names are provided, return all colors
    if not color_names:
        return list(color_dict.values())
    
    # Return the colors that match the provided color names
    result = []
    for name in color_names:
        if name in color_dict:
            result.append(color_dict[name])
        # If the color name is not found, print a warning message
        else:
            print(f"Warning: '{name}' is not a valid color name and will be ignored.")
    
    return result

def create_page_property_2post_parent_db(post_title: str, post_text: str, multi_select_properties: list, child_page_A_ids: list, child_page_B_ids: list) -> dict:
    """
    Create a page property for the parent database.
    """
    page_property = {
        'Name': {
            'id': 'title',
            'type': 'title',
            "title": [{ "type": "text", "text": { "content": post_title } }]
        },
        'テキストフィールド': {
            'id': 'text_field_id',
            'type': 'rich_text',
            'rich_text': [{
                "type": "text",
                "text": {
                    "content": post_text,
                    "link": None
                }
            }],
        },
        'テスト複数セレクト': {
           'multi_select': multi_select_properties
        },
        'リレーション先DB_A': {
            'has_more': False,
            'id': 'relation_column_A_id',
            'relation': [{"id": child_id} for child_id in child_page_A_ids],
            'type': 'relation'
        },
        'リレーション先DB_B': {
            'has_more': False,
            'id': 'relation_column_B_id',
            'relation': [{"id": child_id} for child_id in child_page_B_ids],
            'type': 'relation'
        }
    }
    
    return page_property

# Example usage:
# page_prop = create_page_property("Post Title", "Post Text", [{"name": "Tag1"}, {"name": "Tag2"}], ["id1", "id2"], ["id3", "id4"])