Kralot/sdk-notion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Finaliza ajustes para iniciar Repositorio Git do SDK de Integração do Notion separado do meu projeto original

Browse Source 
---

- Cria singleton de `client` com func `configure` para inicializar e `get_instance` para buscar instância do client;
- Ajusta clients para buscar headers vindo do pai `client` e fixa versão legacy no client de databases;
- Adiciona inicialização de `client` no init do projeto com api_token e api_version informados pelo usuário;
- Altera `NotionConfig` para inserir `database_id` no lugar de `database_name`;
- Altera sistema para receber `database_id` no lugar de `database_name`;
- Altera tipo de `properties` em `schemas.responses.pages.Page` de `Union[Dict[str, Any]], TDB` para `Union[Any, TDB]` para resolver reclamações de type hint;
- Adiciona param `generic_response` no init de `client` e nos clients e databases e pages para pular uso de mapping ao usar `.generic`;
- Adiciona param `raw_response` para pular parser e mappings e retornar resposta original da api;
- Finaliza `types` com subpastas para importações mas com init mãe vazio para evitar dependência circular e permitir uso de `notion.types.` pelo usuário;
- Remove importações do projeto original não relacionadas com o SDK;
- Adiciona param `timezone` na func `start_date` em `orm.common.SetProperty` que antes vinha do env, para posteriormente puxar da init da integração;
- Monta `LICENSE`, `README.md` e `pyproject.toml` base simples para commit inicial do projeto permitindo build de pacote;

---
This commit is contained in:
Eduardo Riguetto
2026-01-20 22:26:39 -03:00
commit aff0446458
72 changed files with 2910 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
notion/orm/__init__.py
Executable
+15
View File
@@ -0,0 +1,15 @@
from .common import Common as _common
from .parsers import Parser as _parser
from .mapping import Mapping as _mapping
from .repositories import repo as _repositories
class _NotionOrm:
def __init__(self) -> None:
self.common = _common
self.parser = _parser
self.mapping = _mapping
self.repo = _repositories
NotionOrm = _NotionOrm()
__all__ = ["NotionOrm"]
notion/orm/accessors/PageProperty.py
Executable
+106
View File
@@ -0,0 +1,106 @@
from typing import Any, Optional
from pydantic import validate_call
from ..extrators.Properties import PropertyExtractor
class _PageProperty(PropertyExtractor):
"Getter específico para propriedades individuais do Notion. Permite buscar uma propriedade específica por nome e tipo."
def _get_properties_dict(self, response: dict) -> dict:
"Extrai o dicionário de properties do response"
nt_dict = response.get('properties')
if nt_dict is None:
nt_dict = response.get('result')
if nt_dict is None:
raise KeyError("Response inserido é inválido")
return nt_dict
def _get_property_data(self, response: dict, name: str, expected_type: str) -> dict:
"""
Busca e valida uma propriedade específica.
Args:
response: Response completo do Notion
name: Nome da propriedade
expected_type: Tipo esperado da propriedade
Returns:
Dados da propriedade
Raises:
KeyError: Se propriedade não existe
ValueError: Se tipo não corresponde
"""
properties = self._get_properties_dict(response)
prop = properties.get(name)
if not prop:
raise KeyError(f"Propriedade '{name}' não foi encontrada")
actual_type = prop.get('type')
if actual_type != expected_type:
raise ValueError(
f"Propriedade '{name}' não é do tipo '{expected_type}' "
f"(tipo atual: '{actual_type}')"
)
return prop
# ==================== CLASSES INTERNAS ====================
@validate_call
def id(self, response: dict) -> str:
prop = response.get('id')
if not prop:
raise KeyError("ID da página não foi encontrada")
return prop
@validate_call
def title(self, response: dict, name: str) -> Optional[str]:
prop = self._get_property_data(response, name, "title")
return self._title(prop)
@validate_call
def text(self, response: dict, name: str) -> Optional[dict]:
prop = self._get_property_data(response, name, "rich_text")
return self._rich_text(prop)
@validate_call
def number(self, response: dict, name: str) -> Optional[float]:
prop = self._get_property_data(response, name, "number")
return self._number(prop)
@validate_call
def select(self, response: dict, name: str) -> Optional[dict]:
prop = self._get_property_data(response, name, "select")
return self._select(prop)
@validate_call
def checkbox(self, response: dict, name: str) -> bool:
prop = self._get_property_data(response, name, "checkbox")
return self._checkbox(prop)
@validate_call
def date(self, response: dict, name: str) -> Optional[dict]:
prop = self._get_property_data(response, name, "date")
return self._date(prop)
@validate_call
def relation(self, response: dict, name: str) -> Optional[list]:
prop = self._get_property_data(response, name, "relation")
return self._relation(prop)
@validate_call
def rollup(self, response: dict, name: str) -> Any:
prop = self._get_property_data(response, name, "rollup")
return self._rollup(prop)
@validate_call
def formula(self, response: dict, name: str) -> Any:
prop = self._get_property_data(response, name, "formula")
return self._formula(prop)
PageProperty = _PageProperty()
__all__ = ['_PageProperty']
notion/orm/accessors/__init__.py
Executable
+9
View File
@@ -0,0 +1,9 @@
from .PageProperty import PageProperty as _PageProperty
class _Acessors:
def __init__(self) -> None:
self.PageProperty = _PageProperty
Acessors = _Acessors
__all__ = ["Acessors"]
notion/orm/common/.examples/QueryFilter.py
Executable
+12
View File
@@ -0,0 +1,12 @@
import json
from ..QueryFilter import QueryFilter
filter1 = QueryFilter.and_(
QueryFilter.title("Name", "contains", "Teste"),
QueryFilter.or_(
QueryFilter.number("Value", "greater_than", 10),
QueryFilter.number("Value", "less_than", 5)
)
)
print(json.dumps({"filter":filter1.to_dict()}))
notion/orm/common/.examples/QuerySort.py
Executable
+12
View File
@@ -0,0 +1,12 @@
import json
from ..QuerySort import QuerySort
sort1 = QuerySort.ascending("Teste")
print(json.dumps({"sorts":sort1.to_dict()}))
sort2 = QuerySort.and_(
QuerySort.ascending("Teste")
)
print(json.dumps({"sorts":sort2.to_dict()}))
notion/orm/common/QueryFilter.py
Executable
+541
View File
@@ -0,0 +1,541 @@
from typing import Any, Dict, Literal, Union
from datetime import datetime, date
class _NotionFilter:
"Classe base para filtros do Notion"
def to_dict(self) -> Dict[str, Any]:
"Converte o filtro para o formato JSON do Notion"
raise NotImplementedError
def and_(self, *filters: '_NotionFilter') -> '_NotionFilter':
"Combina este filtro com outros usando AND"
return _AndFilter(self, *filters)
def or_(self, *filters: '_NotionFilter') -> '_NotionFilter':
"Combina este filtro com outros usando OR"
return _OrFilter(self, *filters)
class _PropertyFilter(_NotionFilter):
"Filtro para uma propriedade específica"
def __init__(self,
property_name: str,
property_type: str,
condition: str,
value: Any
):
self.property_name = property_name
self.property_type = property_type
self.condition = condition
self.value = value
def to_dict(self) -> Dict[str, Any]:
return {
"property": self.property_name,
self.property_type: {
self.condition: self.value
}
}
class _AndFilter(_NotionFilter):
"Combina múltiplos filtros com AND"
def __init__(self, *filters: _NotionFilter):
self.filters = list(filters)
def to_dict(self) -> Dict[str, Any]:
return {
"and": [filter_obj.to_dict() for filter_obj in self.filters]
}
class _OrFilter(_NotionFilter):
"Combina múltiplos filtros com OR"
def __init__(self, *filters: _NotionFilter):
self.filters = list(filters)
def to_dict(self) -> Dict[str, Any]:
return {
"or": [filter_obj.to_dict() for filter_obj in self.filters]
}
class QueryFilter:
"Builder principal para criar filtros"
@staticmethod
def and_(*filters: _NotionFilter) -> _AndFilter:
"Combina filtros com AND"
return _AndFilter(*filters)
@staticmethod
def or_(*filters: _NotionFilter) -> _OrFilter:
"Combina filtros com OR"
return _OrFilter(*filters)
@staticmethod
def created_time(
property_name: str,
condition: Literal[
"equals",
"before",
"after",
"on_or_before",
"on_or_after",
"is_empty",
"is_not_empty"
],
value: Union[date, Literal[True]]
):
if isinstance(value, date):
property_value = value.strftime("%Y-%m-%d")
else:
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "created_time",
condition = condition,
value = property_value
)
@staticmethod
def last_edited_time(
property_name: str,
condition: Literal[
"equals",
"before",
"after",
"on_or_before",
"on_or_after",
"is_empty",
"is_not_empty"
],
value: Union[date, Literal[True]]
):
if isinstance(value, date):
property_value = value.strftime("%Y-%m-%d")
else:
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "last_edited_time",
condition = condition,
value = property_value
)
@staticmethod
def title(
property_name: str,
condition: Literal[
"equals",
"does_not_equal",
"contains",
"does_not_contain",
"starts_with",
"ends_with",
"is_empty",
"is_not_empty"
],
value: str
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "title",
condition = condition,
value = property_value
)
@staticmethod
def rich_text(
property_name: str,
condition: Literal[
"equals",
"does_not_equal",
"contains",
"does_not_contain",
"starts_with",
"ends_with",
"is_empty",
"is_not_empty"
],
value: str
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "rich_text",
condition = condition,
value = property_value
)
@staticmethod
def number(
property_name: str,
condition: Literal[
"equals",
"does_not_equal",
"greater_than",
"less_than",
"greater_than_or_equal_to",
"less_than_or_equal_to",
"is_empty",
"is_not_empty"
],
value: Union[float, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "number",
condition = condition,
value = property_value
)
@staticmethod
def checkbox(
property_name: str,
condition: Literal[
"equals",
"does_not_equal"
],
value: Literal[True]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "checkbox",
condition = condition,
value = property_value
)
@staticmethod
def select(
property_name: str,
condition: Literal[
"equals",
"does_not_equal",
"is_empty",
"is_not_empty"
],
value: Union[str, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "select",
condition = condition,
value = property_value
)
@staticmethod
def multi_select(
property_name: str,
condition: Literal[
"contains",
"does_not_contain",
"equals",
"does_not_equal",
"is_empty",
"is_not_empty"
],
value: Union[str, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "multi_select",
condition = condition,
value = property_value
)
@staticmethod
def status(
property_name: str,
condition: Literal[
"equals",
"does_not_equal",
"is_empty",
"is_not_empty"
],
value: Union[str, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "status",
condition = condition,
value = property_value
)
@staticmethod
def dates(
property_name: str,
condition: Literal[
"equals",
"does_not_equal",
"after",
"on_or_before",
"on_or_after",
"is_empty",
"is_not_empty"
],
value: Union[date, datetime, Literal[True]]
):
if isinstance(value, date):
property_value = value.strftime("%Y-%m-%d")
elif isinstance(value, datetime):
property_value = value.strftime("%Y-%m-%dT%H:%M:%S")
else:
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "date",
condition = condition,
value = property_value
)
@staticmethod
def people(
property_name: str,
condition: Literal[
"contains",
"does_not_contain",
"is_empty",
"is_not_empty"
],
value: Union[str, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "people",
condition = condition,
value = property_value
)
@staticmethod
def files(
property_name: str,
condition: Literal[
"is_empty",
"is_not_empty"
],
value: Literal[True]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "files",
condition = condition,
value = property_value
)
@staticmethod
def url(
property_name: str,
condition: Literal[
"equals",
"does_not_equal",
"contains",
"does_not_contain",
"starts_with",
"ends_with",
"is_empty",
"is_not_empty"
],
value: Union[str, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "url",
condition = condition,
value = property_value
)
@staticmethod
def email(
property_name: str,
condition: Literal[
"equals",
"does_not_equal",
"contains",
"does_not_contain",
"starts_with",
"ends_with",
"is_empty",
"is_not_empty"
],
value: Union[str, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "email",
condition = condition,
value = property_value
)
@staticmethod
def phone_number(
property_name: str,
condition: Literal[
"equals",
"does_not_equal",
"contains",
"does_not_contain",
"starts_with",
"ends_with",
"is_empty",
"is_not_empty"
],
value: Union[str, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "phone_number",
condition = condition,
value = property_value
)
@staticmethod
def relation(
property_name: str,
condition: Literal[
"contains",
"does_not_contain",
"is_empty",
"is_not_empty"
],
value: Union[str, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "relation",
condition = condition,
value = property_value
)
@staticmethod
def created_by(
property_name: str,
condition: Literal[
"contains",
"does_not_contain",
"is_empty",
"is_not_empty"
],
value: Union[str, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "created_by",
condition = condition,
value = property_value
)
@staticmethod
def last_edited_by(
property_name: str,
condition: Literal[
"contains",
"does_not_contain",
"is_empty",
"is_not_empty"
],
value: Union[str, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "last_edited_by",
condition = condition,
value = property_value
)
@staticmethod
def formula(
property_name: str,
formula_type: Literal["string", "checkbox", "number", "date"],
condition: Literal[
"equals",
"does_not_equal",
"contains",
"does_not_contain",
"starts_with",
"ends_with",
"is_empty",
"is_not_empty",
"greater_than",
"less_than",
"greater_than_or_equal_to",
"less_than_or_equal_to",
"before",
"after",
"on_or_before",
"on_or_after"
],
value: Union[str, bool, float, date, datetime, Literal[True]]
) -> _PropertyFilter:
if isinstance(value, date):
property_value = value.strftime("%Y-%m-%d")
elif isinstance(value, datetime):
property_value = value.strftime("%Y-%m-%dT%H:%M:%S")
else:
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = f"formula.{formula_type}",
condition = condition,
value = property_value
)
@staticmethod
def unique_id(
property_name: str,
condition: Literal[
"equals",
"does_not_equal",
"greater_than",
"less_than",
"greater_than_or_equal_to",
"less_than_or_equal_to",
"is_empty",
"is_not_empty"
],
value: Union[float, Literal[True]]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "unique_id",
condition = condition,
value = property_value
)
@staticmethod
def verification(
property_name: str,
condition: Literal[
"status"
],
value: Literal["verified", "expired", "none", ""]
):
property_value = value
return _PropertyFilter(
property_name = property_name,
property_type = "verification",
condition = condition,
value = property_value
)
__all__ = ["QueryFilter"]
notion/orm/common/QuerySort.py
Executable
+72
View File
@@ -0,0 +1,72 @@
from typing import Any, Dict, Literal, List
class _NotionSort:
"Classe base para classificação do Notion"
def _sort_to_dict(self) -> Dict[str, Any]:
"Converte uma classificação para o formato JSON do Notion"
raise NotImplementedError
def to_dict(self) -> List[Dict[str, Any]]:
"Converte uma ou multiplas classificações para o formato JSON do Notion"
raise NotImplementedError
class _PropertySort(_NotionSort):
"Classificação para uma propriedade específica"
def __init__(self,
property_name: str,
direction: Literal["ascending", "descending"]
):
self.property_name = property_name
self.direction = direction
def _sort_to_dict(self) -> Dict[str, Any]:
return {
"property": self.property_name,
"direction": self.direction
}
def to_dict(self) -> List[Dict[str, Any]]:
return [self._sort_to_dict()]
class _MultiSort(_NotionSort):
"Combina múltiplas classificações com vírgula"
def __init__(self, *sorts: _NotionSort):
self.sorts = sorts
def to_dict(self) -> List[Dict[str, Any]]:
return [sort_obj._sort_to_dict() for sort_obj in self.sorts]
class QuerySort:
"Builder principal para criar filtros"
@staticmethod
def and_(*sorts: _NotionSort) -> _MultiSort:
"Combina classificações com vírgula"
return _MultiSort(*sorts)
@staticmethod
def ascending(
property_name: str
):
return _PropertySort(
property_name = property_name,
direction = "ascending"
)
@staticmethod
def descending(
property_name: str
):
return _PropertySort(
property_name = property_name,
direction = "descending"
)
__all__ = ["QuerySort"]
notion/orm/common/SetProperty.py
Executable
+231
View File
@@ -0,0 +1,231 @@
from typing import Dict, Any, Union, Optional, Literal, List, TYPE_CHECKING
from datetime import datetime, date
from pydantic import validate_call
if TYPE_CHECKING:
from ..repositories.pages.CreatePage import CreatePage
from ..repositories.databases.CreateDatabasePage import CreateDatabasePage
class SetProperty:
def __init__(self,
CreatePageClass : Union['CreatePage', 'CreateDatabasePage'],
properties : Dict[str, Any]
) -> None:
self._CreatePage = CreatePageClass
self._properties = properties
@validate_call
def number(self,
name : str,
value : Optional[float]
):
if value is None:
return self._CreatePage
self._properties[name] = {
"number": value
}
return self._CreatePage
@validate_call
def checkbox(self,
name : str,
value : Optional[bool]
):
if value is None:
return self._CreatePage
self._properties[name] = {
"checkbox": value
}
return self._CreatePage
@validate_call
def start_date(self,
name : str,
value : Optional[Union[date, datetime]],
timezone : str = "Etc/UTC"
):
if value is None:
return self._CreatePage
if type(value) == date:
self._properties[name] = {
"date": {
"start": value.strftime("%Y-%m-%d")
}
}
elif type(value) == datetime:
self._properties[name] = {
"date": {
"start": value.strftime("%Y-%m-%dT%H:%M:%S"),
"time_zone": timezone
}
}
return self._CreatePage
@validate_call
def end_date(self,
name : str,
value : Optional[Union[date, datetime]]
):
if value is None:
return self._CreatePage
if not self._properties.get(name):
raise KeyError("start_date is missing")
date_value = None
if type(value) == date:
date_value = value.strftime("%Y-%m-%d")
elif type(value) == datetime:
date_value = value.strftime("%Y-%m-%dT%H:%M:%S")
self._properties[name]["date"]["end"] = date_value
return self._CreatePage
@validate_call
def relation(self,
name : str,
value : Optional[str]
):
if value is None:
return self._CreatePage
self._properties[name] = {
"relation": [
{
"id": str(value)
}
]
}
return self._CreatePage
@validate_call
def text(self,
name : str,
value : Optional[str]
):
if value is None:
return self._CreatePage
self._properties[name] = {
"rich_text" : [
{
"type":"text",
"text":{
"content": value
}
}
]
}
return self._CreatePage
@validate_call
def rich_text(self,
name : str,
value : Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]
):
if value is None:
return self._CreatePage
value = [value] if isinstance(value, dict) else value
self._properties[name] = {
"rich_text" : value
}
return self._CreatePage
@validate_call
def select(self,
name : str,
value : Optional[str],
color : Optional[Literal[
"default",
"gray",
"brown",
"orange",
"yellow",
"green",
"blue",
"purple",
"pink",
"red"
]] = None
):
if value is None:
return self._CreatePage
if color:
self._properties[name] = {
"select" : {
"name" : value,
"color" : color
}
}
return self._CreatePage
self._properties[name] = {
"select" : {
"name" : value
}
}
return self._CreatePage
# @validate_call
# def multi_select(self, name : str, value : str):
# @validate_call
# def status(self, name : str, value : str):
# @validate_call
# def media(self, name : str, value : str):
@validate_call
def url(self,
name : str,
value : Optional[str]
):
if value is None:
return self._CreatePage
self._properties[name] = {
"url" : value
}
return self._CreatePage
@validate_call
def email(self,
name : str,
value : Optional[str]
):
if value is None:
return self._CreatePage
if "@" not in value:
raise ValueError("Invalid email")
self._properties[name] = {
"email": str(value)
}
return self._CreatePage
@validate_call
def phone(self,
name : str,
value : Optional[str]
):
if value is None:
return self._CreatePage
self._properties[name] = {
"phone_number": str(value)
}
return self._CreatePage
@validate_call
def person(self,
name : str,
value : Optional[str]
):
if value is None:
return self._CreatePage
self._properties[name] = {
"people": [
{
"object": "user",
"id": str(value)
}
]
}
return self._CreatePage
# @validate_call
# def place(self, name : str, value : str):
__all__ = ["SetProperty"]
notion/orm/common/__init__.py
Executable
+13
View File
@@ -0,0 +1,13 @@
from .QueryFilter import QueryFilter as _QueryFilter
from .QuerySort import QuerySort as _QuerySort
from .SetProperty import SetProperty as _SetProperty
class _Common:
def __init__(self) -> None:
self.QueryFilter = _QueryFilter
self.QuerySort = _QuerySort
self.SetProperty = _SetProperty
Common = _Common()
__all__ = ["Common"]
notion/orm/extrators/Properties.py
Executable
+146
View File
@@ -0,0 +1,146 @@
from typing import Optional, Any
from datetime import datetime
class PropertyExtractor:
"Classe base com métodos de extração de propriedades do Notion"
def extract(self, prop_data: dict) -> Any:
"""
Extrai o valor de uma propriedade baseado no seu tipo.
Args:
prop_data: Dicionário com dados da propriedade do Notion
Returns:
Valor extraído da propriedade ou None
"""
tipo = prop_data.get("type")
if not tipo:
return None
extractors = {
"title" : self._title,
"rich_text" : self._rich_text,
"number" : self._number,
"checkbox" : self._checkbox,
"url" : self._url,
"select" : self._select,
"multi_select" : self._multi_select,
"date" : self._date,
"relation" : self._relation,
"rollup" : self._rollup,
"formula" : self._formula,
}
extractor = extractors.get(tipo)
if extractor:
return extractor(prop_data)
return None
def _title(self, prop_data: dict) -> Optional[str]:
"Extrai conteúdo de propriedade tipo `title`"
title = prop_data.get("title")
if title is None:
return None
if isinstance(title, list) and len(title) == 1:
return title[0]['text'].get('content')
if isinstance(title, dict):
return title['text'].get('content')
content_list = [item['plain_text'] for item in title]
return "".join(content_list)
def _rich_text(self, prop_data: dict) -> Optional[dict]:
"Extrai conteúdo de propriedade tipo 'rich_text'"
prop_list = prop_data.get('rich_text')
if prop_list is None:
return None
content_list = [item['plain_text'] for item in prop_list]
return {"text": "".join(content_list), "detailed": prop_list}
def _number(self, prop_data: dict) -> Optional[float]:
"Extrai conteúdo de propriedade tipo 'number'"
return prop_data.get("number")
def _checkbox(self, prop_data: dict) -> bool:
"Extrai conteúdo de propriedade tipo 'checkbox'"
return prop_data.get("checkbox") == True
def _url(self, prop_data: dict) -> Optional[str]:
"Extrai conteúdo de propriedade tipo 'url'"
return prop_data.get("url")
def _select(self, prop_data: dict) -> Optional[dict]:
"Extrai conteúdo de propriedade tipo 'select'"
select = prop_data.get("select")
if not select:
return None
return {"name": select.get("name"), "color": select.get("color")}
def _multi_select(self, prop_data: dict) -> Optional[list[dict]]:
"Extrai conteúdo de propriedade tipo 'multi_select'"
multi_select = prop_data.get("multi_select")
if not multi_select:
return None
selects = []
for select in multi_select:
selects.append(
{"name": select.get("name"), "color": select.get("color")}
)
return selects
def _date(self, prop_data: dict) -> Optional[dict]:
"Extrai conteúdo de propriedade tipo 'date'"
date = prop_data.get("date")
if not date:
return None
def parse_date(date_str):
if not date_str:
return None
for fmt in ["%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"]:
try:
return datetime.strptime(date_str, fmt)
except:
continue
return None
start = parse_date(date.get("start"))
end = parse_date(date.get("end"))
return {"start": start, "end": end}
def _relation(self, prop_data: dict) -> Optional[list]:
"Extrai conteúdo de propriedade tipo 'relation'"
relations = prop_data.get("relation", [])
if not relations:
return None
return [relation["id"] for relation in relations]
def _rollup(self, prop_data: dict) -> Any:
"Extrai conteúdo de propriedade tipo 'rollup'"
rollup = prop_data.get("rollup", {})
rollup_type = rollup.get("type")
if rollup_type == "number":
return rollup.get("number")
elif rollup_type == "array":
return rollup.get("array", [])
return None
def _formula(self, prop_data: dict) -> Any:
"Extrai conteúdo de propriedade tipo 'formula'"
formula = prop_data.get("formula", {})
formula_type = formula.get("type")
if formula_type == "number":
return formula.get("number")
elif formula_type == "string":
return formula.get("string")
elif formula_type == "boolean":
return formula.get("boolean")
return None
__all__ = ["PropertyExtractor"]
notion/orm/extrators/__init__.py
Executable
+8
View File
@@ -0,0 +1,8 @@
from .Properties import PropertyExtractor as _PropertyExtractor
class Extrator:
def __init__(self) -> None:
self.props = _PropertyExtractor
__all__ = ["Extrator"]
notion/orm/mapping/__init__.py
Executable
+12
View File
@@ -0,0 +1,12 @@
from .database import NotionDatabase as _NotionDatabase
from .registry import DatabaseRegistry as _DatabaseRegistry
class _Mapping:
NotionDatabase = _NotionDatabase
def __init__(self) -> None:
self.registry = _DatabaseRegistry
Mapping = _Mapping()
__all__ = ["Mapping"]
notion/orm/mapping/database.py
Executable
+198
View File
@@ -0,0 +1,198 @@
from typing import Dict, Any, Optional, Callable, ClassVar, TypeVar, Type
from pydantic import BaseModel, field_validator
T = TypeVar('T', bound='NotionDatabase')
class NotionConfigMeta:
"Metadados de configuração da database do Notion"
def __init__(self, config_class):
self.database_id : Optional[str] = getattr(config_class, 'database_id', None)
self.mappings : Dict[str, Any] = getattr(config_class, 'mappings', {})
self.validators : Dict[str, Callable] = getattr(config_class, 'validators', {})
self.computed : Dict[str, Callable] = getattr(config_class, 'computed', {})
transformers_config: Dict[str, Callable] = getattr(config_class, 'transformers', {})
# Processa mappings para separar nome e transformer
self.field_mappings : Dict[str, str] = {}
self.transformers : Dict[str, Callable] = {}
for field_name, mapping in self.mappings.items():
if isinstance(mapping, str):
# Mapping simples: "field": "Notion Name"
self.field_mappings[field_name] = mapping
elif isinstance(mapping, tuple) and len(mapping) == 2:
# Mapping com transformer: "field": ("Notion Name", lambda x: ...)
self.field_mappings[field_name] = mapping[0]
self.transformers[field_name] = mapping[1]
else:
raise ValueError(
f"Mapping inválido para '{field_name}'. "
f"Use: 'Notion Name' ou ('Notion Name', transformer_func)"
)
self.transformers.update(transformers_config)
class NotionDatabaseMeta(type(BaseModel)):
"Metaclass que processa a classe e injeta funcionalidades do Notion"
_notion_config: NotionConfigMeta
def __new__(mcs, name, bases, namespace, **kwargs):
# Cria a classe Pydantic
cls = super().__new__(mcs, name, bases, namespace, **kwargs)
# Se tem NotionConfig, processa
if hasattr(cls, 'NotionConfig'):
config = NotionConfigMeta(cls.NotionConfig)
# Injeta o config processado
cls._notion_config = config # type: ignore
if not config.database_id:
raise AttributeError("Database ID is missing")
# Injeta validators customizados do NotionConfig
if config.validators:
for field_name, validator_func in config.validators.items():
# Cria um validator Pydantic dinâmico
validator_name = f'validate_{field_name}_notion'
def make_validator(func):
def validator(cls, v):
return func(v)
return field_validator(field_name)(validator)
setattr(cls, validator_name, make_validator(validator_func))
return cls
class NotionDatabase(BaseModel, metaclass = NotionDatabaseMeta):
"""
Classe base para schemas de databases do Notion.
Uso:
----
class AccountsDB(NotionDatabase):
name: str
credit: float
type: List[str]
class NotionConfig:
mappings = {
"name": "Name",
"credit": "Credit",
"type": ("Type", lambda x: [s["name"] for s in x] if x else [])
}
"""
_notion_config: ClassVar[NotionConfigMeta]
@classmethod
def from_notion_page(
cls : Type[T],
page_properties : Dict[str, Any],
page_parser : Optional[Callable] = None
) -> Optional[T]:
"""
Cria uma instância do schema a partir de propriedades parseadas do Notion.
Args:
page_properties: Dict com propriedades já parseadas pelo PropertyExtractor
page_parser: Função de parsing (se as propriedades ainda não foram parseadas)
Returns:
Instância do schema ou None se parsing falhar
"""
if not hasattr(cls, '_notion_config'):
raise AttributeError(
f"{cls.__name__} não tem NotionConfig definido. "
f"Adicione uma subclasse NotionConfig com os mappings."
)
config: NotionConfigMeta = cls._notion_config
# Se recebeu a página raw, faz o parsing
if page_parser and 'properties' in page_properties:
page_properties = page_parser(page_properties)
if not page_properties:
return None
result = {}
# Processa campos mapeados
for field_name, notion_name in config.field_mappings.items():
raw_value = page_properties.get(notion_name)
# Aplica transformer se existir
if field_name in config.transformers:
value = config.transformers[field_name](raw_value)
else:
value = raw_value
result[field_name] = value
# Cria instância temporária para computed fields
instance = cls(**result)
# Processa computed fields
for field_name, compute_func in config.computed.items():
computed_value = compute_func(instance)
result[field_name] = computed_value
# Retorna instância final com computed fields
return cls(**result)
@classmethod
def get_database_id(cls) -> str:
"Retorna o nome da database no Notion"
if hasattr(cls, '_notion_config'):
return cls._notion_config.database_id or cls.__name__
return cls.__name__
@classmethod
def get_notion_field_name(cls,
python_field : str
) -> Optional[str]:
"Retorna o nome do campo no Notion a partir do nome Python"
if hasattr(cls, '_notion_config'):
return cls._notion_config.field_mappings.get(python_field)
return None
@classmethod
def get_all_mappings(cls) -> Dict[str, str]:
"Retorna todos os mappings field_python -> field_notion"
if hasattr(cls, '_notion_config'):
return cls._notion_config.field_mappings.copy()
return {}
__all__ = ["NotionDatabase"]
notion/orm/mapping/registry.py
Executable
+92
View File
@@ -0,0 +1,92 @@
from typing import Dict, Any, Optional, Callable, Type
from .database import NotionDatabase as _NotionDatabase
class DatabaseRegistry:
"""
Registry global de databases do Notion.
Permite acessar schemas e criar parsers dinamicamente.
"""
_databases: Dict[str, Type[_NotionDatabase]] = {}
@classmethod
def register(cls,
database_class : Type[_NotionDatabase]
) -> None:
"Registra uma database no registry"
db_name = database_class.get_database_id()
cls._databases[db_name] = database_class
@classmethod
def get(cls,
database_id : str
) -> Optional[Type[_NotionDatabase]]:
"Retorna a classe de schema de uma database"
return cls._databases.get(database_id)
@classmethod
def get_parser(cls,
database_id : str,
page_parser : Callable
) -> Callable:
"""
Retorna uma função parser para a database.
Args:
database_id: ID da Database
page_parser: Função que parseia páginas (ex: PageProperties.parse)
Returns:
Função parser ou None se database não encontrada
"""
db_class = cls.get(database_id)
if not db_class:
raise ValueError(f"Database de ID '{database_id}' não está registrada")
def parser(page : Dict[str, Any]) -> Optional[_NotionDatabase]:
return db_class.from_notion_page(page, page_parser)
return parser
@classmethod
def list_databases(cls) -> list[str]:
"Lista todas as databases registradas"
return list(cls._databases.keys())
@classmethod
def auto_register(cls,
*database_classes : Type[_NotionDatabase]
) -> None:
"Registra múltiplas databases de uma vez"
for db_class in database_classes:
cls.register(db_class)
@classmethod
def generate_literal_type(cls) -> str:
"Gera o código do Literal com todas as databases registradas"
db_names = list(cls._databases.keys())
literal_str = ", ".join([f'"{name}"' for name in db_names])
return f"Literal[{literal_str}]"
@classmethod
def get_database_class_by_id(cls,
db_id : str
) -> Optional[Type[_NotionDatabase]]:
"Retorna a classe schema baseado no nome da database"
return cls._databases.get(db_id)
notion/orm/parsers/PageProperties.py
Executable
+36
View File
@@ -0,0 +1,36 @@
from typing import Optional, Dict, Any
from urllib.parse import unquote
from ..extrators.Properties import PropertyExtractor as _PropertyExtractor
class _PageProperties(_PropertyExtractor):
"Parser completo de página do Notion. Retorna todas as propriedades parseadas de uma vez."
def parse(self, page: Dict[str, Any]) -> Optional[Dict[str, Any]]:
result = {}
if page.get("object") == "property_item":
value = self.extract(page)
if value is not None:
result[unquote(page["id"])] = value
return result
properties = page.get("properties")
if not properties:
return None
for prop_name, prop_data in properties.items():
value = self.extract(prop_data)
if value is not None:
result[prop_name] = value
return result
PageProperties = _PageProperties()
__all__ = ["PageProperties"]
notion/orm/parsers/__init__.py
Executable
+9
View File
@@ -0,0 +1,9 @@
from .PageProperties import PageProperties as _PageProperties
class _Parser:
def __init__(self) -> None:
self.page_props = _PageProperties.parse
Parser = _Parser()
__all__ = ["Parser"]
notion/orm/repositories/__init__.py
Executable
+26
View File
@@ -0,0 +1,26 @@
from typing import TypeVar, Generic, TYPE_CHECKING
if TYPE_CHECKING:
from ...schemas.orm.database.DatabasesContainer import DatabasesContainer as _DatabasesContainer
TContainer = TypeVar('TContainer', bound = '_DatabasesContainer')
class DatabasesRepo(Generic[TContainer]):
def __init__(self):
self.container: TContainer
@staticmethod
def generic(database_id : str):
from .databases import Database
return Database.generic(database_id)
class _Repositories(Generic[TContainer]):
def __init__(self):
from .pages import _Pages
self.pages = _Pages()
self.databases: DatabasesRepo[TContainer] = DatabasesRepo()
repo: _Repositories = _Repositories()
__all__ = ["repo", "_Repositories"]
notion/orm/repositories/databases/CreateDatabasePage.py
Executable
+93
View File
@@ -0,0 +1,93 @@
from typing import Optional, Literal, Generic, TypeVar
from pydantic import validate_call
from ....schemas.responses import Schemas as _schm
from ...mapping.database import NotionDatabase as _NotionDatabase
from ...mapping import Mapping as _map
from ...parsers import Parser as _parser
from ...common.SetProperty import SetProperty as _setProperty
from ..pages.CreatePage import CreatePage as _CreatePage
TDB = TypeVar('TDB', bound = _NotionDatabase)
class CreateDatabasePage(Generic[TDB]):
def __init__(self,
database_id : str,
generic_response : bool = False
) -> None:
self._database_id = database_id
self._generic_response = generic_response
self._instance = _CreatePage().set_parent(
type = "database_id",
parent_id = self._database_id
)
self.set_property = _setProperty(self, self._instance.data["properties"])
@validate_call
def set_template(self,
type : Literal["default", "template_id"],
template_id : Optional[str] = None
):
self._instance.set_template(
type = type,
template_id = template_id
)
return self
@validate_call
def set_title(self,
prop_name : Optional[str],
prop_value : Optional[str]
):
self._instance.set_title(
prop_name = prop_name,
prop_value = prop_value
)
return self
@validate_call
def set_icon(self,
type : Literal["external"],
content : Optional[str] = None
):
self._instance.set_icon(
type = type,
content = content
)
return self
@validate_call
async def call(self,
map_properties : bool = True,
raw_response : bool = False
) -> _schm.pages.Page[TDB]:
page = await self._instance.call(
parse_properties = False
)
if not raw_response:
if map_properties:
parser = None
if not self._generic_response:
# Tenta pegar parser do registry
parser = _map.registry.get_parser(
database_id = self._database_id,
page_parser = _parser.page_props
)
# Fallback: se database não registrada, usa parser genérico
if not parser:
parser = _parser.page_props
page = page.model_dump()
mapped_properties = parser(page = page)
page["properties"] = mapped_properties
page = _schm.pages.Page(**page)
return page
__all__ = ["CreateDatabasePage"]
notion/orm/repositories/databases/SearchPage.py
Executable
+101
View File
@@ -0,0 +1,101 @@
from typing import Optional, Generic, TypeVar
from pydantic import validate_call
from ....client import get_client as _get_client
from ....schemas.responses import Schemas as _schm
from ...mapping.database import NotionDatabase as _NotionDatabase
from ...mapping import Mapping as _map
from ...common.QueryFilter import QueryFilter as _Filter, _NotionFilter
from ...common.QuerySort import QuerySort as _Sort, _NotionSort
from ...parsers import Parser as _parser
TDB = TypeVar('TDB', bound = _NotionDatabase)
class SearchPage(Generic[TDB]):
def __init__(self,
database_id : str,
generic_response : bool = False
) -> None:
self._database_id = database_id
self._generic_response = generic_response
self.query_limit : int = 100
self.filter = _Filter
self._filter_obj : Optional[_NotionFilter] = None
self.sort = _Sort
self._sort_obj : Optional[_NotionSort] = None
def set_limit(self,
page_limit : int
):
"Define o limite de páginas a serem retornadas na requisição"
self.query_limit = page_limit
return self
def set_sort(self,
sort_obj : _NotionSort
):
"Define a classificação a ser usada na query"
self._sort_obj = sort_obj
return self
def set_filter(self,
filter_obj : _NotionFilter
):
"Define o filtro a ser usado na query"
self._filter_obj = filter_obj
return self
@validate_call
async def call(self,
map_properties : bool = True,
raw_response : bool = False
) -> _schm.databases.Query[TDB]:
payload = {}
if self.query_limit:
payload["page_size"] = self.query_limit
if self._sort_obj:
payload["sorts"] = self._sort_obj.to_dict()
if self._filter_obj:
payload["filter"] = self._filter_obj.to_dict()
client = _get_client()
query = await client.databases.query(
database_id = self._database_id,
json_data = payload
)
if query['object'] == 'error':
error = _schm.errors.Error(**query)
raise KeyError(error.__dict__)
pages = []
results = query["results"]
if not raw_response:
if map_properties:
parser = None
if not self._generic_response:
# Tenta pegar parser do registry
parser = _map.registry.get_parser(
database_id = self._database_id,
page_parser = _parser.page_props
)
# Fallback: se database não registrada, usa parser genérico
if not parser:
parser = _parser.page_props
for page in results:
page_properties = parser(page)
page["properties"] = page_properties if page_properties else None
pages.append(page)
results = pages
return _schm.databases.Query(**query)
__all__ = ["SearchPage"]
notion/orm/repositories/databases/SearchPageProperty.py
Executable
+94
View File
@@ -0,0 +1,94 @@
from typing import Optional, Literal, List
from ....client import get_client as _get_client
from ....schemas.orm.database import Schemas as _schm
from ....schemas.responses import Schemas as _schmResponses
from ...common.QueryFilter import QueryFilter as _Filter, _NotionFilter
from ...parsers import Parser as _parser
class SearchPageProperty:
"""
# CUIDADO AO UTILIZAR!!
### [Bug Conhecido](https://community.latenode.com/t/notion-api-relation-property-showing-empty-array-despite-ui-showing-connected-pages/25780) na API do Notion impede retornos confiáveis em propriedades paginadas
Ultimo teste : 2026-01-10
"""
def __init__(self, database_id : str) -> None:
self._database_id = database_id
self.filter = _Filter
self._filter_obj : Optional[_NotionFilter] = None
self._property_type = None
def set_filter(self,
property_type : Literal[
"unique_id",
"icon",
"title",
"number",
"checkbox",
"start_date",
"end_date",
#"relation",
"text",
"rich_text",
"select",
"multi_select",
"status",
"media",
"url",
"email",
"phone",
"person",
"place"
],
filter_obj : _NotionFilter
):
"Define o filtro a ser usado na query"
self._property_type = property_type
self._filter_obj = filter_obj
return self
async def call(self,
raw_response : bool = False
) -> List[_schm.SearchPageProperty]:
if not self._property_type:
raise TypeError("Filter is not fully defined")
payload = {}
if self._filter_obj:
payload["filter"] = self._filter_obj.to_dict()
client = _get_client()
query = await client.databases.query_propriety(
database_id = self._database_id,
propriety_type = self._property_type,
json_data = payload
)
if query['object'] == 'error':
error = _schmResponses.errors.Error(**query)
raise KeyError(error.__dict__)
results = query["results"]
pages = []
for page in results:
if raw_response:
page["properties"] = _parser.page_props(page = page)
pages.append(page)
result = []
for page in pages:
prop_value = page["properties"]
if prop_value:
prop_value = list(prop_value.values())[0]
result.append(
_schm.SearchPageProperty(
page_id = page["id"],
prop_value = prop_value
)
)
return result
notion/orm/repositories/databases/__init__.py
Executable
+57
View File
@@ -0,0 +1,57 @@
from typing import Generic, TypeVar
from ...mapping.database import NotionDatabase as _NotionDatabase
from ..pages import _Pages as _Pages
from .CreateDatabasePage import CreateDatabasePage as _CreateDatabasePage
from .SearchPage import SearchPage as _SearchPage
from .SearchPageProperty import SearchPageProperty as _SearchPageProperty
TDB = TypeVar('TDB', bound = _NotionDatabase)
class Database(Generic[TDB]):
def __init__(self,
database_id : str,
generic_response : bool = False
) -> None:
self._database_id = database_id
self._generic_response = generic_response
@property
def CreateDatabasePage(self) -> _CreateDatabasePage[TDB]:
return _CreateDatabasePage(
database_id = self._database_id,
generic_response = self._generic_response
)
@property
def SearchPage(self) -> _SearchPage[TDB]:
return _SearchPage(
database_id = self._database_id,
generic_response = self._generic_response
)
@property
def SearchPageProperty(self) -> _SearchPageProperty:
return _SearchPageProperty(
database_id = self._database_id
)
@property
def page(self) -> _Pages[TDB]:
return _Pages(
database_id = self._database_id,
generic_response = self._generic_response
)
@staticmethod
def generic(database_id : str) -> 'Database[_NotionDatabase]':
"Acessa database sem schema definido"
return Database(
database_id = database_id,
generic_response = True
)
__all__ = ["Database"]
notion/orm/repositories/pages/.examples/CreatePage.py
Executable
+8
View File
@@ -0,0 +1,8 @@
# async def main():
# instance = NotionOrm.repo.pages()
# create = await instance.CreatePage()\
# .set_parent("page_id", "2a564c9be67881a185c1c5d9133b9b1c")\
# .set_title("Name", "Teste abc")\
# .set_children("heading_1", "Teste")\
# .call()
# return create
notion/orm/repositories/pages/.examples/GetPage.py
Executable
+22
View File
@@ -0,0 +1,22 @@
import sys, os, asyncio
sys.path.append(
os.path.abspath(
os.path.join(
os.path.dirname(__file__), '..', '..', '..', '..', '..', '..', '..'
)
)
)
from src.utils.pprint import pprint
from src.integrations.notion.orm.repositories.pages.GetPage import GetPage
async def main():
instance = GetPage()
search = await instance\
.set_database(name="accounts")\
.set_pageid("0db2806f-b365-4327-919d-afbd1943f2ad")\
.select("Name")
#.call(True)
return search
test = await main()
pprint(test)
notion/orm/repositories/pages/.examples/GetPageProperty.py
Executable
+21
View File
@@ -0,0 +1,21 @@
import sys, os, asyncio
sys.path.append(
os.path.abspath(
os.path.join(
os.path.dirname(__file__), '..', '..', '..', '..', '..', '..', '..'
)
)
)
from src.utils.pprint import pprint
from src.integrations.notion.orm.repositories.pages.GetPageProperty import GetPageProperty
async def main():
instance = GetPageProperty()
search = await instance\
.set_pageid("0db2806f-b365-4327-919d-afbd1943f2ad")\
.set_propname("Movements")\
.call()
return search
test = await main()
pprint(test)
notion/orm/repositories/pages/.examples/__init__.py
Executable
View File
notion/orm/repositories/pages/CreatePage.py
Executable
+176
View File
@@ -0,0 +1,176 @@
from typing import Literal, Any, Optional
from pydantic import validate_call
from ....schemas.responses.pages.Page import Page as _schmPage
from ....schemas.responses.errors.Error import Error as _schmError
from ....client import get_client as _get_client
from ...common.SetProperty import SetProperty as _setProperty
from ...parsers import Parser as _parser
class CreatePage:
def __init__(self) -> None:
self.data : dict[str, Any] = {
"properties": {}
}
self.set_property = _setProperty(self, self.data["properties"])
@validate_call
def set_parent(self,
type : Literal["page_id", "database_id"],
parent_id : str
):
self.data["parent"] = {
type : parent_id
}
return self
@validate_call
def set_template(self,
type : Literal["default", "template_id"],
template_id : Optional[str] = None
):
if type == "template_id":
if template_id is None:
raise TypeError("Template Type was set to 'template_id' but no ID was provided")
self.data["template"] = {
"type": type,
"template_id": template_id
}
return self
self.data["template"] = {
"type": type
}
return self
@validate_call
def set_title(self,
prop_name : Optional[str],
prop_value : Optional[str]
):
if prop_value is None:
return self
if self.data["parent"].get("page_id"):
self.data["properties"] = {
"title": [
{
"text": {
"content": prop_value
}
}
]
}
return self
self.data["properties"][prop_name] = {
"title": [
{
"text": {
"content": prop_value
}
}
]
}
return self
@validate_call
def set_icon(self,
type: Literal["external"],
content: Optional[str]
):
if content is None:
return self
match type:
case "external":
self.data["icon"] = {
"external": {
"url": str(content)
}
}
return self
@validate_call
def set_children(self,
type: Literal["heading_1", "paragraph"],
content: Any
):
if self.data.get("children") is None:
self.data["children"] = []
match type:
case "heading_1":
self.data["children"].append(
{
"object": "block",
"type": "heading_1",
"heading_1": {
"rich_text": [
{
"type": "text",
"text": {
"content": content
}
}
]
}
}
)
case "paragraph":
self.data["children"].append(
{
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [
{
"type": "text",
"text": {
"content": content
}
}
]
}
}
)
return self
async def call(self,
parse_properties : bool = True
) -> _schmPage:
client = _get_client()
create = await client.pages.create(
json_data = self.data
)
if create['object'] == 'error':
error = _schmError(**create)
raise KeyError(error.__dict__)
if parse_properties:
properties = _parser.page_props(page = create)
create["properties"] = properties
return _schmPage(**create)
__all__ = ["CreatePage"]
notion/orm/repositories/pages/GetPage.py
Executable
+92
View File
@@ -0,0 +1,92 @@
from typing import Dict, Any, Optional, TypeVar, Generic
from ....schemas.responses.pages.Page import Page as _schmPage
from ....schemas.responses.errors.Error import Error as _schmError
from ....client import get_client as _get_client
from ...mapping.database import NotionDatabase as _NotionDatabase
from ...mapping import Mapping as _map
from ...parsers import Parser as _parser
TDB = TypeVar('TDB', bound =_NotionDatabase)
class GetPage(Generic[TDB]):
def __init__(self,
database_id : Optional[str] = None,
generic_response : bool = False
) -> None:
self._database_id : Optional[str] = database_id
self._generic_response = generic_response
self._pageid : str
def set_pageid(self,
id : str
) -> 'GetPage[TDB]':
self._pageid = id
return self
def set_database(self,
id : str
) -> 'GetPage[TDB]':
self._database_id = id
return self
async def select(self,
property : str
) -> Optional[Any]:
client = _get_client()
page : Dict[str, Any] = await client.pages.get(
page_id = self._pageid
)
if page['object'] == 'error':
error = _schmError(**page)
raise KeyError(error.__dict__)
page["properties"] = _parser.page_props(page = page)
if page["properties"] is None:
return None
allprops = page["properties"]
return allprops[property]
async def call(self,
map_properties : bool = True,
raw_response : bool = False
) -> _schmPage[TDB]:
client = _get_client()
page : Dict[str, Any] = await client.pages.get(
page_id = self._pageid
)
if page['object'] == 'error':
error = _schmError(**page)
raise KeyError(error.__dict__)
if not raw_response:
if map_properties and self._database_id:
parser = None
if not self._generic_response:
# Tenta pegar parser do registry
parser = _map.registry.get_parser(
database_id = self._database_id,
page_parser = _parser.page_props
)
# Fallback: se database não registrada, usa parser genérico
if not parser:
parser = _parser.page_props
mapped_properties = parser(page = page)
page["properties"] = mapped_properties
elif map_properties:
page["properties"] = _parser.page_props(page = page)
return _schmPage(**page)
notion/orm/repositories/pages/GetPageProperty.py
Executable
+63
View File
@@ -0,0 +1,63 @@
from pydantic import validate_call
from typing import Dict, Any, Optional
from ....schemas.responses.errors.Error import Error as _schmError
from ....client import get_client as _get_client
from ...parsers.PageProperties import PageProperties as _PageProperties
class GetPageProperty:
"""
# CUIDADO AO UTILIZAR!!
### [Bug Conhecido](https://community.latenode.com/t/notion-api-relation-property-showing-empty-array-despite-ui-showing-connected-pages/25780) na API do Notion impede retornos confiáveis em propriedades paginadas
Ultimo teste : 2026-01-10
"""
def __init__(self) -> None:
self._pageid : str
self._propname : str
@validate_call
def set_pageid(self,
id : str
):
self._pageid = id
return self
@validate_call
def set_propname(self,
name : str
):
self._propname = name
return self
async def call(self,
raw_response : bool = False
) -> Optional[Dict[str, Any]]:
client = _get_client()
getprop = await client.pages.get_property(
page_id = self._pageid,
property_name = self._propname
)
if getprop['object'] == 'error':
error = _schmError(**getprop)
raise KeyError(error.__dict__)
prop_data = getprop.get("results")
if getprop.get("object") == "property_item":
prop_data = [getprop]
if not prop_data:
return None
if not raw_response:
result = _PageProperties.parse(
page = prop_data[0]
)
result = prop_data[0]
return result
notion/orm/repositories/pages/__init__.py
Executable
+30
View File
@@ -0,0 +1,30 @@
from typing import Optional, Generic, TypeVar
from ...mapping.database import NotionDatabase as _NotionDatabase
from .CreatePage import CreatePage as _CreatePage
from .GetPage import GetPage as _GetPage
from .GetPageProperty import GetPageProperty as _GetPageProperty
TDB = TypeVar('TDB', bound = _NotionDatabase)
class _Pages(Generic[TDB]):
def __init__(self,
database_id : Optional[str] = None,
generic_response : bool = False
) -> None:
self._database_id = database_id
self._generic_response = generic_response
self.CreatePage = _CreatePage
self.GetPageProperty = _GetPageProperty
@property
def GetPage(self) -> _GetPage:
return _GetPage(
database_id = self._database_id,
generic_response = self._generic_response
)
Pages = _Pages()
__all__ = ["Pages", "_Pages"]