Skip to content

deployment

Module diagram

classDiagram
  class deployment {
  }
  class package_version_updater {
  }

deployment

package_version_updater

PACKAGE_VERSION_DATABASE_ID module-attribute 

PACKAGE_VERSION_DATABASE_ID = '1494418ac858801c9129cb7a1dd90e0c'

convert_data_into_exact_match_filters 

convert_data_into_exact_match_filters(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:
    {
        "filter":
            "and": [
                {
                    "property": "exp_name",
                    "text": {"equals": "cls_base"}
                },
                {
                    "property": "accuracy",
                    "number": {"equals": 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/deployment/package_version_updater.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def convert_data_into_exact_match_filters(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:
            {
                "filter":
                    "and": [
                        {
                            "property": "exp_name",
                            "text": {"equals": "cls_base"}
                        },
                        {
                            "property": "accuracy",
                            "number": {"equals": 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
    """
    filters = []
    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

        available_filter_ops = get_avaliable_filter_ops(property_.name)
        if "equals" in available_filter_ops:
            filter_ops = "equals"
        elif "contains" in available_filter_ops:
            filter_ops = "contains"
        elif "is_not empty" in available_filter_ops:
            filter_ops = "is_not_empty"
        else:
            print(f"filter for property {property_.name} not supported")
            continue

        if isinstance(value, list):
            _filters = []
            for _v in value:
                _filters.append(property_.get_filter(key, filter_ops, _v))
            filters.append(Filter.compound("and", _filters))
        else:
            filters.append(property_.get_filter(key, filter_ops, value))

    return Filter.compound("and", filters).to_dict()

query_page 

query_page(data: Dict[str, Any], database_id: str = PACKAGE_VERSION_DATABASE_ID) -> Optional[List]

Queries the Notion database for the provided data.

Parameters:

  • data (Dict[str, Any]) –

    Dictionary containing the data to be queried in the Notion database.

  • database_id (str, default: PACKAGE_VERSION_DATABASE_ID ) –

    The ID of the Notion database.

Source code in ResearchToolkit/deployment/package_version_updater.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def query_page(data: Dict[str, Any], database_id: str = PACKAGE_VERSION_DATABASE_ID) -> Optional[List]:
    """Queries the Notion database for the provided data.

    Args:
        data (Dict[str, Any]): Dictionary containing the data to be queried in the Notion database.
        database_id (str): The ID of the Notion database.
    """

    database_fields = retrieve_database_fields(database_id=database_id)
    filters = convert_data_into_exact_match_filters(data=data, fields=database_fields)
    try:
        _, _, pages = query_database(database_id=database_id, request_body=filters)
    except RuntimeError as e:
        if "select option" in str(e) and "not found" in str(e):
            return []
        raise
    return pages

create_or_load_page 

create_or_load_page(data: Dict[str, Any], database_id: str = PACKAGE_VERSION_DATABASE_ID) -> str

Queries the Notion database for the provided data, and if not found, creates a new page in the Notion database with the provided data.

Parameters:

  • data (Dict[str, Any]) –

    Dictionary containing the data to be added to the Notion database.

Returns:

  • str –

    Optional[str]: The ID of the created page.

Source code in ResearchToolkit/deployment/package_version_updater.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def create_or_load_page(data: Dict[str, Any], database_id: str = PACKAGE_VERSION_DATABASE_ID) -> str:
    """Queries the Notion database for the provided data, and
    if not found, creates a new page in the Notion database with the provided data.

    Args:
        data (Dict[str, Any]): Dictionary containing the data to be added to the Notion database.

    Returns:
        Optional[str]: The ID of the created page.
    """
    page_candidates = query_page(data, database_id=database_id)
    if len(page_candidates) > 1:
        raise ValueError(f"Multiple pages found with the same data: {data}")
    elif len(page_candidates) == 1:
        page_id = page_candidates[0]["id"]
    else:
        _, _, page_id = add_data_to_database(database_id=database_id, data=data)
    return page_id

get_latest_version 

get_latest_version(package_name: str, database_id: str = PACKAGE_VERSION_DATABASE_ID) -> Optional[str]

Queries the Notion database for the latest version of the package.

Parameters:

  • package_name (str) –

    The name of the package.

Returns:

  • Optional[str] –

    Optional[str]: The latest version of the package.

Source code in ResearchToolkit/deployment/package_version_updater.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def get_latest_version(
    package_name: str, database_id: str = PACKAGE_VERSION_DATABASE_ID
) -> Optional[str]:
    """Queries the Notion database for the latest version of the package.

    Args:
        package_name (str): The name of the package.

    Returns:
        Optional[str]: The latest version of the package.
    """
    data = {"Package Name": package_name}
    page_candidates = query_page(data, database_id=database_id)
    if not page_candidates:
        return None

    versions = []
    for page in page_candidates:
        _value = get_property_value(page["properties"]["Version"])
        if _value is None:
            continue
        version = Version.from_string(_value)
        versions.append(version)

    if not versions:
        version = None
    else:
        version = max(versions) if len(versions) > 0 else versions[0]
    return version