From e188d4d94d5ad65eed11165153e1c68b93a86829 Mon Sep 17 00:00:00 2001 From: Ted Kaemming <65315+tkaemming@users.noreply.github.com> Date: Fri, 2 Aug 2024 20:48:04 -0700 Subject: [PATCH] improve type aliases --- posthog/clickhouse/property_groups.py | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/posthog/clickhouse/property_groups.py b/posthog/clickhouse/property_groups.py index 652bb0e9a1f1d5..68bda6d044e8d6 100644 --- a/posthog/clickhouse/property_groups.py +++ b/posthog/clickhouse/property_groups.py @@ -17,8 +17,8 @@ def contains(self, property_key: str) -> bool: return self.key_filter_function(property_key) -Table = str -Column = str +TableName = str +ColumnName = str PropertyGroupName = str @@ -26,24 +26,26 @@ class PropertyGroupManager: def __init__( self, cluster: str, - groups: Mapping[Table, Mapping[Column, Mapping[PropertyGroupName, PropertyGroupDefinition]]], + groups: Mapping[TableName, Mapping[ColumnName, Mapping[PropertyGroupName, PropertyGroupDefinition]]], ) -> None: self.__cluster = cluster self.__groups = groups - def __get_map_column_name(self, column: Column, group_name: PropertyGroupName) -> str: + def __get_map_column_name(self, column: ColumnName, group_name: PropertyGroupName) -> str: return f"{column}_group_{group_name}" - def __get_map_expression(self, column: Column, definition: PropertyGroupDefinition) -> str: + def __get_map_expression(self, column: ColumnName, definition: PropertyGroupDefinition) -> str: return f"mapSort(mapFilter((key, _) -> {definition.key_filter_expression}, CAST(JSONExtractKeysAndValues({column}, 'String'), 'Map(String, String)')))" - def get_property_group_columns(self, table: Table, column: Column, property_key: str) -> Iterable[str]: + def get_property_group_columns(self, table: TableName, column: ColumnName, property_key: str) -> Iterable[str]: if (table_groups := self.__groups.get(table)) and (column_groups := table_groups.get(column)): for group_name, group_definition in column_groups.items(): if group_definition.contains(property_key): yield self.__get_map_column_name(column, group_name) - def get_alter_table_statements(self, table: Table, column: Column, group_name: PropertyGroupName) -> Iterable[str]: + def get_alter_table_statements( + self, table: TableName, column: ColumnName, group_name: PropertyGroupName + ) -> Iterable[str]: group_definition = self.__groups[table][column][group_name] map_column_name = self.__get_map_column_name(column, group_name) column_definition = (