bedrock.generators.code_generators.endpoint.endpoint_class
1from bedrock._helpers.string import kebab_case_to_camelCase, upper_case_first 2from bedrock.generators.code_generators.endpoint.helpers import _resolve_endpoint_method_vars, _resolve_path_param_vars, \ 3 ensure_last_path_param_var_is_in_body, ensure_parent_object_exists 4from bedrock.generators.code_generators.common.helpers import get_model_class_name_from_endpoint 5from bedrock.generators.code_generators.common.imports import import_models 6from bedrock.generators.introspection.decorators import get_class_decorators 7 8 9def make_endpoint_class(kebab_name: str, kafka_topics: list[str], model: str, protected: str, 10 prefixes: list[str], app_directory: str) -> str: 11 name = kebab_case_to_camelCase(kebab_name) 12 parent_class = "EndpointWithKafkaSync" if len(kafka_topics) > 0 else "Endpoint" 13 lines = [] 14 if len(kafka_topics) > 0: 15 kafka_topics_list_string = '["' + '", "'.join(kafka_topics) + '"]' 16 lines.append(f"@kafka_listener({kafka_topics_list_string})") 17 endpoint_prefix = f"prefix=\"/{'/'.join(prefixes)}\", " if prefixes else "" 18 lines.append(f"class {upper_case_first(name)}({parent_class}):") 19 lines.append(f' def __init__(self):') 20 lines.append(f' super().__init__("/{kebab_name}/", {endpoint_prefix}related_model={model})') 21 methods = [make_endpoint__get_single, 22 make_endpoint__get_global, 23 make_endpoint__post_global, 24 make_endpoint__put_single, 25 make_endpoint__delete_single] 26 for make_endpoint_method in methods: 27 lines.append("") 28 lines.extend(make_endpoint_method(kebab_name, protected, prefixes, app_directory, related_model=model, indent=4)) 29 return "\n".join(lines) 30 31 32def get_filter_path_hint_attributes(protected: str, app_directory: str, related_model_name: str) -> str or None: 33 models = import_models(app_directory) 34 protector_class_name = get_model_class_name_from_endpoint(protected) 35 related_model = models.get(related_model_name) 36 model_decorators = get_class_decorators(related_model) 37 filter_path_should_be_self = related_model_name == protector_class_name 38 protector_model_decorator = [decorator for decorator in model_decorators if 39 decorator.name == "@filter_path_hint" and ( 40 protector_class_name in decorator.args[0] or \ 41 filter_path_should_be_self and "self" in decorator.args[0] 42 )] 43 if protector_model_decorator: 44 return protector_model_decorator[0].args[0].split(".")[1].strip("'").strip('"') 45 return None 46 47 48def make_endpoint__common(method_name: str, kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 49 related_model_name: str, indent: int, annotations: list[str] = [], 50 include_default_filter: bool = True) -> list[str]: 51 protector_attribute = get_filter_path_hint_attributes(protected, app_directory, related_model_name) 52 protector_annotation = f"@protected('{protected}', keys=get_keys('{kebab_name}'))" 53 protector_api_annotation = f"@with_query_param('api_{protected}[in]', str, 'api_{protected}', True)" 54 _annotations = [*annotations] 55 lines = [] 56 extra_variables = [] 57 if protected: 58 _annotations.append(protector_annotation) 59 extra_variables = [f"{protected}=[]"] 60 if protected and "@filter_on_columns()" in _annotations: 61 _annotations.append(protector_api_annotation) 62 if protector_api_annotation in _annotations: 63 extra_variables.append(f"api_{protected}=[]") 64 65 lines.extend(_annotations) 66 if "@paginated()" in annotations: 67 extra_variables.extend([ 68 'sort_order="desc"', 69 'sort_column="created_at"', 70 'offset=None', 71 'limit=None' 72 ]) 73 endpoint_variables = ', '.join(_resolve_endpoint_method_vars(extra_variables)) 74 lines.append(f"def {method_name}({endpoint_variables}) -> tuple[int, BedrockResponse]:") 75 if include_default_filter: 76 _protector_attribute = "chase_account_number" if protected == "accounts" else protector_attribute 77 if _protector_attribute: 78 lines.append(f' default_filter = {{"{_protector_attribute}[in]": {protected}}}') 79 else: 80 lines.append(" default_filter = {}") 81 if protector_annotation in _annotations and protector_api_annotation in _annotations: 82 p = protected[0] 83 lines.append(f' if "tkc_token" in event:') 84 lines.append(f' default_filter["{_protector_attribute}[in]"] = [{p} for {p} in api_{protected} if {p} in {protected}] if api_{protected} != [] else {protected}') 85 lines.append(f' if "tkc_token" not in event and api_{protected} != []:') 86 lines.append(f' default_filter["{_protector_attribute}[in]"] = api_{protected}') 87 88 if protected and not protector_attribute: 89 print("Warning: No filter path hint found for the protected attribute. This may cause issues with filtering.") 90 lines.append( 91 " raise NotImplementedError(\"No default filter for this endpoint. This is likely due to this endpoint having been generated automatically but the automation was unable to figure out how to use the default filter.\")") 92 93 lines.extend(_resolve_path_param_vars(prefixes, indent=indent)) 94 return lines 95 96 97def make_endpoint__get_single(kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 98 related_model=None, indent: int = 4) -> list[str]: 99 annotations = [f"@with_query_param('api_{protected}[in]', str, 'api_{protected}', True)"] if protected else [] 100 lines = make_endpoint__common("get_single", kebab_name, protected, prefixes, app_directory, 101 related_model, indent, annotations) 102 lines.append(" return self.get_single_generic(event, belongs_to=default_filter)") 103 return [f"{' ' * indent}{line}" for line in lines] 104 105 106def make_endpoint__get_global(kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 107 related_model=None, indent: int = 4) -> list[str]: 108 lines = make_endpoint__common("get_global", kebab_name, protected, prefixes, app_directory, related_model, indent, 109 ["@paginated()", "@filter_on_columns()"]) 110 lines.append(" return self.get_global_generic(event,") 111 lines.append(" belongs_to={**filters, **default_filter},") 112 lines.append(" allow_listing_all='tkc_token' not in event,") 113 lines.append(" order_by=sort_column,") 114 lines.append(" order=sort_order,") 115 lines.append(" offset=offset,") 116 lines.append(" limit=limit)") 117 return [f"{' ' * indent}{line}" for line in lines] 118 119 120def make_endpoint__post_global(kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 121 related_model=None, indent: int = 4) -> list[str]: 122 has_parent_objects = any("{" in prefix for prefix in prefixes) 123 annotations = [f"@with_query_param('api_{protected}[in]', str, 'api_{protected}', True)"] if protected and has_parent_objects else [] 124 lines = make_endpoint__common("post_global", kebab_name, protected, prefixes, app_directory, 125 related_model, indent, annotations, 126 include_default_filter=has_parent_objects) 127 lines.append(" body = json.loads(event[\"body\"])") 128 lines.append(ensure_last_path_param_var_is_in_body(prefixes, indent=indent)) 129 lines.extend(ensure_parent_object_exists(prefixes, protected, indent=indent)) 130 lines.append(" return self.post_global_generic(event, altered_body=body)") 131 return [f"{' ' * indent}{line}" for line in lines] 132 133 134def make_endpoint__put_single(kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 135 related_model=None, indent: int = 4) -> list[str]: 136 annotations = [f"@with_query_param('api_{protected}[in]', str, 'api_{protected}', True)"] if protected else [] 137 lines = make_endpoint__common("put_single", kebab_name, protected, prefixes, app_directory, 138 related_model, indent, annotations) 139 lines.append(" body = json.loads(event[\"body\"])") 140 lines.append(" body[\"uuid\"] = event[\"pathParameters\"][self.param_key]") 141 lines.append(ensure_last_path_param_var_is_in_body(prefixes, indent=indent)) 142 lines.extend(ensure_parent_object_exists(prefixes, protected, indent=indent)) 143 lines.append(" return self.put_single_generic(event, altered_body=body, belongs_to=default_filter)") 144 return [f"{' ' * indent}{line}" for line in lines] 145 146 147def make_endpoint__delete_single(kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 148 related_model=None, indent: int = 4) -> list[str]: 149 annotations = [f"@with_query_param('api_{protected}[in]', str, 'api_{protected}', True)"] if protected else [] 150 lines = make_endpoint__common("delete_single", kebab_name, protected, prefixes, app_directory, 151 related_model, indent, annotations) 152 lines.append(" uuid = event[\"pathParameters\"][self.param_key]") 153 lines.extend(ensure_parent_object_exists(prefixes, protected, indent=indent)) 154 lines.append(" return self.delete_single_generic(event, resource_id=uuid, belongs_to=default_filter)") 155 return [f"{' ' * indent}{line}" for line in lines]
def
make_endpoint_class( kebab_name: str, kafka_topics: list[str], model: str, protected: str, prefixes: list[str], app_directory: str) -> str:
10def make_endpoint_class(kebab_name: str, kafka_topics: list[str], model: str, protected: str, 11 prefixes: list[str], app_directory: str) -> str: 12 name = kebab_case_to_camelCase(kebab_name) 13 parent_class = "EndpointWithKafkaSync" if len(kafka_topics) > 0 else "Endpoint" 14 lines = [] 15 if len(kafka_topics) > 0: 16 kafka_topics_list_string = '["' + '", "'.join(kafka_topics) + '"]' 17 lines.append(f"@kafka_listener({kafka_topics_list_string})") 18 endpoint_prefix = f"prefix=\"/{'/'.join(prefixes)}\", " if prefixes else "" 19 lines.append(f"class {upper_case_first(name)}({parent_class}):") 20 lines.append(f' def __init__(self):') 21 lines.append(f' super().__init__("/{kebab_name}/", {endpoint_prefix}related_model={model})') 22 methods = [make_endpoint__get_single, 23 make_endpoint__get_global, 24 make_endpoint__post_global, 25 make_endpoint__put_single, 26 make_endpoint__delete_single] 27 for make_endpoint_method in methods: 28 lines.append("") 29 lines.extend(make_endpoint_method(kebab_name, protected, prefixes, app_directory, related_model=model, indent=4)) 30 return "\n".join(lines)
def
get_filter_path_hint_attributes(protected: str, app_directory: str, related_model_name: str) -> str:
33def get_filter_path_hint_attributes(protected: str, app_directory: str, related_model_name: str) -> str or None: 34 models = import_models(app_directory) 35 protector_class_name = get_model_class_name_from_endpoint(protected) 36 related_model = models.get(related_model_name) 37 model_decorators = get_class_decorators(related_model) 38 filter_path_should_be_self = related_model_name == protector_class_name 39 protector_model_decorator = [decorator for decorator in model_decorators if 40 decorator.name == "@filter_path_hint" and ( 41 protector_class_name in decorator.args[0] or \ 42 filter_path_should_be_self and "self" in decorator.args[0] 43 )] 44 if protector_model_decorator: 45 return protector_model_decorator[0].args[0].split(".")[1].strip("'").strip('"') 46 return None
def
make_endpoint__common( method_name: str, kebab_name: str, protected: str, prefixes: list[str], app_directory: str, related_model_name: str, indent: int, annotations: list[str] = [], include_default_filter: bool = True) -> list[str]:
49def make_endpoint__common(method_name: str, kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 50 related_model_name: str, indent: int, annotations: list[str] = [], 51 include_default_filter: bool = True) -> list[str]: 52 protector_attribute = get_filter_path_hint_attributes(protected, app_directory, related_model_name) 53 protector_annotation = f"@protected('{protected}', keys=get_keys('{kebab_name}'))" 54 protector_api_annotation = f"@with_query_param('api_{protected}[in]', str, 'api_{protected}', True)" 55 _annotations = [*annotations] 56 lines = [] 57 extra_variables = [] 58 if protected: 59 _annotations.append(protector_annotation) 60 extra_variables = [f"{protected}=[]"] 61 if protected and "@filter_on_columns()" in _annotations: 62 _annotations.append(protector_api_annotation) 63 if protector_api_annotation in _annotations: 64 extra_variables.append(f"api_{protected}=[]") 65 66 lines.extend(_annotations) 67 if "@paginated()" in annotations: 68 extra_variables.extend([ 69 'sort_order="desc"', 70 'sort_column="created_at"', 71 'offset=None', 72 'limit=None' 73 ]) 74 endpoint_variables = ', '.join(_resolve_endpoint_method_vars(extra_variables)) 75 lines.append(f"def {method_name}({endpoint_variables}) -> tuple[int, BedrockResponse]:") 76 if include_default_filter: 77 _protector_attribute = "chase_account_number" if protected == "accounts" else protector_attribute 78 if _protector_attribute: 79 lines.append(f' default_filter = {{"{_protector_attribute}[in]": {protected}}}') 80 else: 81 lines.append(" default_filter = {}") 82 if protector_annotation in _annotations and protector_api_annotation in _annotations: 83 p = protected[0] 84 lines.append(f' if "tkc_token" in event:') 85 lines.append(f' default_filter["{_protector_attribute}[in]"] = [{p} for {p} in api_{protected} if {p} in {protected}] if api_{protected} != [] else {protected}') 86 lines.append(f' if "tkc_token" not in event and api_{protected} != []:') 87 lines.append(f' default_filter["{_protector_attribute}[in]"] = api_{protected}') 88 89 if protected and not protector_attribute: 90 print("Warning: No filter path hint found for the protected attribute. This may cause issues with filtering.") 91 lines.append( 92 " raise NotImplementedError(\"No default filter for this endpoint. This is likely due to this endpoint having been generated automatically but the automation was unable to figure out how to use the default filter.\")") 93 94 lines.extend(_resolve_path_param_vars(prefixes, indent=indent)) 95 return lines
def
make_endpoint__get_single( kebab_name: str, protected: str, prefixes: list[str], app_directory: str, related_model=None, indent: int = 4) -> list[str]:
98def make_endpoint__get_single(kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 99 related_model=None, indent: int = 4) -> list[str]: 100 annotations = [f"@with_query_param('api_{protected}[in]', str, 'api_{protected}', True)"] if protected else [] 101 lines = make_endpoint__common("get_single", kebab_name, protected, prefixes, app_directory, 102 related_model, indent, annotations) 103 lines.append(" return self.get_single_generic(event, belongs_to=default_filter)") 104 return [f"{' ' * indent}{line}" for line in lines]
def
make_endpoint__get_global( kebab_name: str, protected: str, prefixes: list[str], app_directory: str, related_model=None, indent: int = 4) -> list[str]:
107def make_endpoint__get_global(kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 108 related_model=None, indent: int = 4) -> list[str]: 109 lines = make_endpoint__common("get_global", kebab_name, protected, prefixes, app_directory, related_model, indent, 110 ["@paginated()", "@filter_on_columns()"]) 111 lines.append(" return self.get_global_generic(event,") 112 lines.append(" belongs_to={**filters, **default_filter},") 113 lines.append(" allow_listing_all='tkc_token' not in event,") 114 lines.append(" order_by=sort_column,") 115 lines.append(" order=sort_order,") 116 lines.append(" offset=offset,") 117 lines.append(" limit=limit)") 118 return [f"{' ' * indent}{line}" for line in lines]
def
make_endpoint__post_global( kebab_name: str, protected: str, prefixes: list[str], app_directory: str, related_model=None, indent: int = 4) -> list[str]:
121def make_endpoint__post_global(kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 122 related_model=None, indent: int = 4) -> list[str]: 123 has_parent_objects = any("{" in prefix for prefix in prefixes) 124 annotations = [f"@with_query_param('api_{protected}[in]', str, 'api_{protected}', True)"] if protected and has_parent_objects else [] 125 lines = make_endpoint__common("post_global", kebab_name, protected, prefixes, app_directory, 126 related_model, indent, annotations, 127 include_default_filter=has_parent_objects) 128 lines.append(" body = json.loads(event[\"body\"])") 129 lines.append(ensure_last_path_param_var_is_in_body(prefixes, indent=indent)) 130 lines.extend(ensure_parent_object_exists(prefixes, protected, indent=indent)) 131 lines.append(" return self.post_global_generic(event, altered_body=body)") 132 return [f"{' ' * indent}{line}" for line in lines]
def
make_endpoint__put_single( kebab_name: str, protected: str, prefixes: list[str], app_directory: str, related_model=None, indent: int = 4) -> list[str]:
135def make_endpoint__put_single(kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 136 related_model=None, indent: int = 4) -> list[str]: 137 annotations = [f"@with_query_param('api_{protected}[in]', str, 'api_{protected}', True)"] if protected else [] 138 lines = make_endpoint__common("put_single", kebab_name, protected, prefixes, app_directory, 139 related_model, indent, annotations) 140 lines.append(" body = json.loads(event[\"body\"])") 141 lines.append(" body[\"uuid\"] = event[\"pathParameters\"][self.param_key]") 142 lines.append(ensure_last_path_param_var_is_in_body(prefixes, indent=indent)) 143 lines.extend(ensure_parent_object_exists(prefixes, protected, indent=indent)) 144 lines.append(" return self.put_single_generic(event, altered_body=body, belongs_to=default_filter)") 145 return [f"{' ' * indent}{line}" for line in lines]
def
make_endpoint__delete_single( kebab_name: str, protected: str, prefixes: list[str], app_directory: str, related_model=None, indent: int = 4) -> list[str]:
148def make_endpoint__delete_single(kebab_name: str, protected: str, prefixes: list[str], app_directory: str, 149 related_model=None, indent: int = 4) -> list[str]: 150 annotations = [f"@with_query_param('api_{protected}[in]', str, 'api_{protected}', True)"] if protected else [] 151 lines = make_endpoint__common("delete_single", kebab_name, protected, prefixes, app_directory, 152 related_model, indent, annotations) 153 lines.append(" uuid = event[\"pathParameters\"][self.param_key]") 154 lines.extend(ensure_parent_object_exists(prefixes, protected, indent=indent)) 155 lines.append(" return self.delete_single_generic(event, resource_id=uuid, belongs_to=default_filter)") 156 return [f"{' ' * indent}{line}" for line in lines]