👍 Page properties are most useful when interacting with a page that is an entry in a database, represented as a row in the Notion UI. If a page is not part of a database, then its only available property is its title . A page object is made up of page properties that contain data about the page. Wh...
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"]
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')
## 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