From ca7ae468fab2c7c8b83479f15910264e1b3d32ad Mon Sep 17 00:00:00 2001 From: Steve Messick Date: Thu, 9 Jul 2026 21:59:10 +0000 Subject: [PATCH] Fix dropped handler code and reformat --- kagglesdk/competitions/types/hackathons.py | 16 + kagglesdk/kaggle_http_client.py | 88 +-- kagglesdk/kaggle_object.py | 595 +++++++++++---------- 3 files changed, 376 insertions(+), 323 deletions(-) diff --git a/kagglesdk/competitions/types/hackathons.py b/kagglesdk/competitions/types/hackathons.py index ed42a27..839591f 100644 --- a/kagglesdk/competitions/types/hackathons.py +++ b/kagglesdk/competitions/types/hackathons.py @@ -1,3 +1,4 @@ +from datetime import datetime from kagglesdk.competitions.types.competition_enums import HackathonTrackPrizeType from kagglesdk.competitions.types.team import Team from kagglesdk.discussions.types.writeup_types import WriteUp @@ -204,6 +205,7 @@ class HackathonWriteUp(KaggleObject): should_create_doi (bool) Whether or not the user indicated a DOI should be created for the hackathon writeup when the competition closes. + update_time (datetime) """ def __init__(self): @@ -217,6 +219,7 @@ def __init__(self): self._owner_host_user_id = None self._owner_judge_user_id = None self._should_create_doi = None + self._update_time = None self._freeze() @property @@ -357,6 +360,18 @@ def should_create_doi(self, should_create_doi: Optional[bool]): raise TypeError('should_create_doi must be of type bool') self._should_create_doi = should_create_doi + @property + def update_time(self) -> datetime: + return self._update_time or None + + @update_time.setter + def update_time(self, update_time: Optional[datetime]): + if update_time is None: + del self.update_time + return + if not isinstance(update_time, datetime): + raise TypeError('update_time must be of type datetime') + self._update_time = update_time HackathonTrack._fields = [ FieldMetadata("id", "id", "_id", int, 0, PredefinedSerializer()), @@ -386,5 +401,6 @@ def should_create_doi(self, should_create_doi: Optional[bool]): FieldMetadata("ownerHostUserId", "owner_host_user_id", "_owner_host_user_id", int, None, PredefinedSerializer(), optional=True), FieldMetadata("ownerJudgeUserId", "owner_judge_user_id", "_owner_judge_user_id", int, None, PredefinedSerializer(), optional=True), FieldMetadata("shouldCreateDoi", "should_create_doi", "_should_create_doi", bool, None, PredefinedSerializer(), optional=True), + FieldMetadata("updateTime", "update_time", "_update_time", datetime, None, DateTimeSerializer(), optional=True), ] diff --git a/kagglesdk/kaggle_http_client.py b/kagglesdk/kaggle_http_client.py index da33bfe..8ede93b 100644 --- a/kagglesdk/kaggle_http_client.py +++ b/kagglesdk/kaggle_http_client.py @@ -42,10 +42,14 @@ def _get_apikey_creds(): if not kaggle_json or not kaggle_json.strip(): return None - api_key_data = json.loads(kaggle_json) - username = api_key_data["username"] - api_key = api_key_data["key"] - return username, api_key + try: + # Be careful, since the file may be used for more than credential storage. + api_key_data = json.loads(kaggle_json) + username = api_key_data["username"] + api_key = api_key_data["key"] + return username, api_key + except KeyError: + return None class KaggleHttpClient(object): @@ -55,14 +59,14 @@ class KaggleHttpClient(object): _xsrf_header_name = "X-XSRF-TOKEN" def __init__( - self, - env: KaggleEnv = None, - verbose: bool = False, - username: str = None, - password: str = None, - api_token: str = None, - user_agent: str = "kaggle-api/v1.7.0", # Was: V2 - response_processor=None, + self, + env: KaggleEnv = None, + verbose: bool = False, + username: str = None, + password: str = None, + api_token: str = None, + user_agent: str = "kaggle-api/v1.7.0", # Was: V2 + response_processor=None, ): self._env = env or get_env() self._signed_in = None @@ -76,21 +80,25 @@ def __init__( self._response_processor = response_processor def call( - self, - service_name: str, - request_name: str, - request: KaggleObject, - response_type: Type[KaggleObject], + self, + service_name: str, + request_name: str, + request: KaggleObject, + response_type: Type[KaggleObject], ): self._init_session() http_request = self._prepare_request(service_name, request_name, request) # Merge environment settings into session - settings = self._session.merge_environment_settings(http_request.url, {}, None, None, None) + settings = self._session.merge_environment_settings( + http_request.url, {}, None, None, None + ) # Use stream=True for file downloads to avoid loading entire file into memory # See: https://github.com/Kaggle/kaggle-api/issues/754 - if response_type is not None and (response_type == FileDownload or response_type == HttpRedirect): + if response_type is not None and ( + response_type == FileDownload or response_type == HttpRedirect + ): settings["stream"] = True http_response = self._session.send(http_request, **settings) @@ -98,7 +106,9 @@ def call( response = self._prepare_response(response_type, http_response) return response - def _prepare_request(self, service_name: str, request_name: str, request: KaggleObject): + def _prepare_request( + self, service_name: str, request_name: str, request: KaggleObject + ): request_url = self._get_request_url(service_name, request_name) http_request = requests.Request( method="POST", @@ -118,7 +128,9 @@ def _prepare_response(self, response_type, http_response): if "application/json" in http_response.headers["Content-Type"]: resp = http_response.json() if "code" in resp and resp["code"] >= 400: - raise requests.exceptions.HTTPError(resp["message"], response=http_response) + raise requests.exceptions.HTTPError( + resp["message"], response=http_response + ) except KeyError: pass http_response.raise_for_status() @@ -133,7 +145,9 @@ def _print_request(self, request): if not self._verbose: return self._print("---------------------Request----------------------") - self._print(f"{request.method} {request.url}\n{_headers_to_str(request.headers)}\n\n{request.body}") + self._print( + f"{request.method} {request.url}\n{_headers_to_str(request.headers)}\n\n{request.body}" + ) self._print("--------------------------------------------------") def _print_response(self, response, body=True): @@ -162,7 +176,9 @@ def _init_session(self): return self._session self._session = requests.Session() - self._session.headers.update({"User-Agent": self._user_agent, "Content-Type": "application/json"}) + self._session.headers.update( + {"User-Agent": self._user_agent, "Content-Type": "application/json"} + ) iap_token = self._get_iap_token_if_required() if iap_token is not None: @@ -203,17 +219,19 @@ def _fill_xsrf_token(self, iap_token): self._session.headers.update( { - KaggleHttpClient._xsrf_header_name: self._session.cookies[KaggleHttpClient._xsrf_cookie_name], + KaggleHttpClient._xsrf_header_name: self._session.cookies[ + KaggleHttpClient._xsrf_cookie_name + ], } ) def build_start_oauth_url( - self, - client_id: str, - redirect_uri: str, - scope: list[str], - state: str, - code_challenge: str, + self, + client_id: str, + redirect_uri: str, + scope: list[str], + state: str, + code_challenge: str, ) -> str: params = { "response_type": "code", @@ -234,7 +252,9 @@ def get_oauth_default_redirect_url(self) -> str: return f"{self.get_non_api_endpoint()}/account/api/oauth/token" def get_non_api_endpoint(self) -> str: - return "https://www.kaggle.com" if self._env == KaggleEnv.PROD else self._endpoint + return ( + "https://www.kaggle.com" if self._env == KaggleEnv.PROD else self._endpoint + ) class BearerAuth(requests.auth.AuthBase): @@ -250,7 +270,7 @@ def _try_fill_auth(self): return if self._api_token is None: - (api_token, _) = get_access_token_from_env() + api_token, _ = get_access_token_from_env() self._api_token = api_token if self._api_token is not None: @@ -272,5 +292,7 @@ def _try_fill_auth(self): def _get_request_url(self, service_name: str, request_name: str): # On prod, API endpoints are served under https://api.kaggle.com/v1, # but on staging/admin/local, they are served under http://localhost/api/v1. - base_url = self._endpoint if self._env == KaggleEnv.PROD else f"{self._endpoint}/api" + base_url = ( + self._endpoint if self._env == KaggleEnv.PROD else f"{self._endpoint}/api" + ) return f"{base_url}/v1/{service_name}/{request_name}" diff --git a/kagglesdk/kaggle_object.py b/kagglesdk/kaggle_object.py index 299276b..b2b0fc5 100644 --- a/kagglesdk/kaggle_object.py +++ b/kagglesdk/kaggle_object.py @@ -6,342 +6,357 @@ class ObjectSerializer(object): - def __init__(self, to_dict_value, from_dict_value): - self.to_dict_value = to_dict_value - self.from_dict_value = from_dict_value + def __init__(self, to_dict_value, from_dict_value): + self.to_dict_value = to_dict_value + self.from_dict_value = from_dict_value class PredefinedSerializer(ObjectSerializer): - def __init__(self): - """Predefined objects such as int, float etc are serialized/deserialized directly.""" - ObjectSerializer.__init__(self, lambda cls, v, _: v, lambda cls, v: v) + def __init__(self): + """Predefined objects such as int, float etc are serialized/deserialized directly.""" + ObjectSerializer.__init__(self, lambda cls, v, _: v, lambda cls, v: v) # Adapted from https://stackoverflow.com/questions/1175208/elegant-python-function-to-convert-camelcase-to-snake-case -_pascal_to_upper_snake_case_regex = re.compile("((?<=[a-z0-9])[A-Z]|(?!^)[A-Z](?=[a-z]))") +_pascal_to_upper_snake_case_regex = re.compile( + "((?<=[a-z0-9])[A-Z]|(?!^)[A-Z](?=[a-z]))" +) def _pascal_case_to_upper_snake_case(string): - return _pascal_to_upper_snake_case_regex.sub(r"_\1", string).upper() + return _pascal_to_upper_snake_case_regex.sub(r"_\1", string).upper() def _convert(camel_input): - words = re.findall(r"[A-Z]?[a-z]+|[A-Z]{2,}(?=[A-Z][a-z]|\d|\W|$)|\d+", camel_input) - return "_".join(map(str.lower, words)) + words = re.findall(r"[A-Z]?[a-z]+|[A-Z]{2,}(?=[A-Z][a-z]|\d|\W|$)|\d+", camel_input) + return "_".join(map(str.lower, words)) class EnumSerializer(ObjectSerializer): - def __init__(self): - """ - Enum objects are serialized using their ".name" field and deserialized by indexing the string in the Enum type. - Example: - class Foo(Enum): - TEST = 1 - foo = Foo.TEST - foo.name # => returns "TEST" - Foo["TEST"] # => returns Foo.TEST enum value. - """ - ObjectSerializer.__init__( - self, - lambda cls, v, _: EnumSerializer._to_str(cls, v), - lambda cls, v: EnumSerializer._from_str(cls, v), - ) - - @staticmethod - def _to_str(cls, v): - # "v" corresponds to an enum instance: Example foo or Foo.Test above. - # "cls" corresponds to the enum type Foo above. - return v.name - - @staticmethod - def _from_str(cls, v): - # "v" corresponds to enum string: Example "TEST" above. - # "cls" corresponds to the enum type Foo above. - try: - return cls[v] - except KeyError: - dct = vars(cls) - n = v.lower() - nn = _convert(v).lower() - enum_prefix = _pascal_case_to_upper_snake_case(cls.__name__).lower() - for key in dct.keys(): - k = key.lower() - if k == n: - return dct[key] - if k.startswith(enum_prefix) and k.endswith(n) or k.endswith(nn): - return dct[key] - raise + def __init__(self): + """ + Enum objects are serialized using their ".name" field and deserialized by indexing the string in the Enum type. + Example: + class Foo(Enum): + TEST = 1 + foo = Foo.TEST + foo.name # => returns "TEST" + Foo["TEST"] # => returns Foo.TEST enum value. + """ + ObjectSerializer.__init__( + self, + lambda cls, v, _: EnumSerializer._to_str(cls, v), + lambda cls, v: EnumSerializer._from_str(cls, v), + ) + + @staticmethod + def _to_str(cls, v): + # "v" corresponds to an enum instance: Example foo or Foo.Test above. + # "cls" corresponds to the enum type Foo above. + return v.name + + @staticmethod + def _from_str(cls, v): + # "v" corresponds to enum string: Example "TEST" above. + # "cls" corresponds to the enum type Foo above. + try: + return cls[v] + except KeyError: + dct = vars(cls) + n = v.lower() + nn = _convert(v).lower() + enum_prefix = _pascal_case_to_upper_snake_case(cls.__name__).lower() + for key in dct.keys(): + k = key.lower() + if k == n: + return dct[key] + if k.startswith(enum_prefix) and k.endswith(n) or k.endswith(nn): + return dct[key] + raise class ListSerializer(ObjectSerializer): - def __init__(self, item_serializer: ObjectSerializer): - """ - Lists are serialized based on the type they contain. Since objects are generated from proto files, a list always - contains objects of the same type, which is serialized/deserialized using "item_serializer". - """ - ObjectSerializer.__init__( - self, - lambda cls, l, ignore_defaults: [item_serializer.to_dict_value(cls, v, ignore_defaults) for v in l], - lambda cls, l: [item_serializer.from_dict_value(cls, v) for v in l], - ) + def __init__(self, item_serializer: ObjectSerializer): + """ + Lists are serialized based on the type they contain. Since objects are generated from proto files, a list always + contains objects of the same type, which is serialized/deserialized using "item_serializer". + """ + ObjectSerializer.__init__( + self, + lambda cls, l, ignore_defaults: [ + item_serializer.to_dict_value(cls, v, ignore_defaults) for v in l + ], + lambda cls, l: [item_serializer.from_dict_value(cls, v) for v in l], + ) class MapSerializer(ObjectSerializer): - def __init__(self, item_serializer: ObjectSerializer): - """ - Maps are serialized based on type of their values. Since maps keys are always predefined types, we don't need a - serializer for them. - """ - ObjectSerializer.__init__( - self, - lambda cls, d, ignore_defaults: { - k: item_serializer.to_dict_value(cls, v, ignore_defaults) for k, v in d.items() - }, - lambda cls, d: {k: item_serializer.from_dict_value(cls, v) for k, v in d.items()}, - ) + def __init__(self, item_serializer: ObjectSerializer): + """ + Maps are serialized based on type of their values. Since maps keys are always predefined types, we don't need a + serializer for them. + """ + ObjectSerializer.__init__( + self, + lambda cls, d, ignore_defaults: { + k: item_serializer.to_dict_value(cls, v, ignore_defaults) + for k, v in d.items() + }, + lambda cls, d: { + k: item_serializer.from_dict_value(cls, v) for k, v in d.items() + }, + ) class DateTimeSerializer(ObjectSerializer): - def __init__(self): - """Date times are serialized/deserialized as a string in iso format""" - ObjectSerializer.__init__( - self, - lambda cls, dt, _: DateTimeSerializer._to_str(dt), - lambda _, v: DateTimeSerializer._from_str(v), - ) - - @staticmethod - def _to_str(dt): - return dt.isoformat(timespec="milliseconds") + "Z" - - @staticmethod - def _from_str(v): - v = v.rstrip("Z") - fields = v.rsplit(".", maxsplit=1) - if len(fields) == 1: - return datetime.fromisoformat(v) - (dt, nanos) = fields - millis = nanos[:3] - try: - return datetime.fromisoformat(f"{dt}.{millis}") - except ValueError: - return datetime.fromisoformat(dt) # Python 3.9, 3.10 + def __init__(self): + """Date times are serialized/deserialized as a string in iso format""" + ObjectSerializer.__init__( + self, + lambda cls, dt, _: DateTimeSerializer._to_str(dt), + lambda _, v: DateTimeSerializer._from_str(v), + ) + + @staticmethod + def _to_str(dt): + return dt.isoformat(timespec="milliseconds") + "Z" + + @staticmethod + def _from_str(v): + v = v.rstrip("Z") + fields = v.rsplit(".", maxsplit=1) + if len(fields) == 1: + return datetime.fromisoformat(v) + (dt, nanos) = fields + millis = nanos[:3] + try: + return datetime.fromisoformat(f"{dt}.{millis}") + except ValueError: + return datetime.fromisoformat(dt) # Python 3.9, 3.10 class TimeDeltaSerializer(ObjectSerializer): - # Scaling factors idea from https://github.com/protocolbuffers/protobuf/blob/master/csharp/src/Google.Protobuf/JsonParser.cs - SUBSECOND_SCALING_FACTORS = [ - 0, - 100_000_000, - 10_000_000, - 1_000_000, - 100_000, - 10_000, - 1_000, - 100, - 10, - 1, - ] - - def __init__(self): - """Time deltas are serialized/deserialized as a string in ".s" format. Example: 151.500s""" - ObjectSerializer.__init__( - self, - lambda cls, t, _: TimeDeltaSerializer._to_dict_value(t), - lambda cls, v: TimeDeltaSerializer._from_dict_value(v), - ) - - @staticmethod - def _to_dict_value(delta): - seconds = delta.seconds - nanos_str = TimeDeltaSerializer._nanos_to_str(delta.microseconds * 1000) - if nanos_str is None: - return "{}s".format(seconds) - return "{}.{}s".format(seconds, nanos_str) - - @staticmethod - def _nanos_to_str(nanos): - if nanos == 0: - return None - if nanos < 0: - nanos *= -1 - if nanos % 1000000 == 0: - return "{}".format(nanos / 1000000) - elif nanos % 1000 == 0: - return "{}".format(nanos / 1000) - else: - return "{}".format(nanos) - - @staticmethod - def _from_dict_value(value): - raw_value = value.rstrip("s") - if "." not in raw_value: - return timedelta(seconds=int(raw_value)) - seconds, nanosRaw = raw_value.split(".", 1) - nanos = int(nanosRaw) * TimeDeltaSerializer.SUBSECOND_SCALING_FACTORS[len(nanosRaw)] - return timedelta(seconds=int(seconds), microseconds=int(int(nanos) / 1000)) + # Scaling factors idea from https://github.com/protocolbuffers/protobuf/blob/master/csharp/src/Google.Protobuf/JsonParser.cs + SUBSECOND_SCALING_FACTORS = [ + 0, + 100_000_000, + 10_000_000, + 1_000_000, + 100_000, + 10_000, + 1_000, + 100, + 10, + 1, + ] + + def __init__(self): + """Time deltas are serialized/deserialized as a string in ".s" format. Example: 151.500s""" + ObjectSerializer.__init__( + self, + lambda cls, t, _: TimeDeltaSerializer._to_dict_value(t), + lambda cls, v: TimeDeltaSerializer._from_dict_value(v), + ) + + @staticmethod + def _to_dict_value(delta): + seconds = delta.seconds + nanos_str = TimeDeltaSerializer._nanos_to_str(delta.microseconds * 1000) + if nanos_str is None: + return "{}s".format(seconds) + return "{}.{}s".format(seconds, nanos_str) + + @staticmethod + def _nanos_to_str(nanos): + if nanos == 0: + return None + if nanos < 0: + nanos *= -1 + if nanos % 1000000 == 0: + return "{}".format(nanos / 1000000) + elif nanos % 1000 == 0: + return "{}".format(nanos / 1000) + else: + return "{}".format(nanos) + + @staticmethod + def _from_dict_value(value): + raw_value = value.rstrip("s") + if "." not in raw_value: + return timedelta(seconds=int(raw_value)) + seconds, nanosRaw = raw_value.split(".", 1) + nanos = int(nanosRaw) * TimeDeltaSerializer.SUBSECOND_SCALING_FACTORS[len(nanosRaw)] + return timedelta(seconds=int(seconds), microseconds=int(int(nanos) / 1000)) class FieldMaskSerializer(ObjectSerializer): - def __init__(self): - """Field masks are serialized/deserialized as a string that contains a list of paths with a comma delimiter""" - ObjectSerializer.__init__( - self, - lambda cls, m, _: m.ToJsonString(), - lambda cls, v: FieldMaskSerializer._from_joined_paths(v), - ) + def __init__(self): + """Field masks are serialized/deserialized as a string that contains a list of paths with a comma delimiter""" + ObjectSerializer.__init__( + self, + lambda cls, m, _: m.ToJsonString(), + lambda cls, v: FieldMaskSerializer._from_joined_paths(v), + ) - @staticmethod - def _from_joined_paths(joined_paths): - mask = FieldMask() - mask.FromJsonString(joined_paths) - return mask + @staticmethod + def _from_joined_paths(joined_paths): + mask = FieldMask() + mask.FromJsonString(joined_paths) + return mask class KaggleObjectSerializer(ObjectSerializer): - def __init__(self): - """ - Kaggle objects (i.e., proto-generated types that inherit from KaggleObject) have custom "to_dict" and "from_dict" - methods that serialize/deserialize them to/from dictionaries. - """ - ObjectSerializer.__init__( - self, - # "v" is an instance of a KaggleObject. For example: "req = ListCompetitionsRequest()". - # So "req.to_dict()" returns a dictionary with keys as json field names. Example: - # '{"pageSize": 10, "page": 2}' - lambda cls, v, ignore_defaults: cls.to_dict(v, ignore_defaults), - # "cls" is the type of a KaggleObject. For example: ListCompetitionsRequest. All - # generated Kaggle objects have "from_dict" class method that takes a dict to create a - # new instance of the object. See "KaggleObject" class definition below. - lambda cls, v: cls.from_dict(v), - ) + def __init__(self): + """ + Kaggle objects (i.e., proto-generated types that inherit from KaggleObject) have custom "to_dict" and "from_dict" + methods that serialize/deserialize them to/from dictionaries. + """ + ObjectSerializer.__init__( + self, + # "v" is an instance of a KaggleObject. For example: "req = ListCompetitionsRequest()". + # So "req.to_dict()" returns a dictionary with keys as json field names. Example: + # '{"pageSize": 10, "page": 2}' + lambda cls, v, ignore_defaults: cls.to_dict(v, ignore_defaults), + # "cls" is the type of a KaggleObject. For example: ListCompetitionsRequest. All + # generated Kaggle objects have "from_dict" class method that takes a dict to create a + # new instance of the object. See "KaggleObject" class definition below. + lambda cls, v: cls.from_dict(v), + ) class FieldMetadata(object): - def __init__( - self, - json_name, - field_name, - private_field_name, - field_type, - default_value, - serializer, - optional=False, - ): - self.json_name = json_name - self.field_name = field_name - self.private_field_name = private_field_name - self.field_type = field_type - self.default_value = default_value - self.serializer = serializer - self.optional = optional - - def get_as_dict_item(self, instance, ignore_defaults=True): - value = getattr(instance, self.private_field_name) - if ignore_defaults and value == self.default_value: - return None - if value is None: - return None - return self.serializer.to_dict_value(self.field_type, value, ignore_defaults) - - def set_from_dict(self, instance, json_dict): - if self.json_name not in json_dict: - return # Ignore unknown fields - value = json_dict[self.json_name] - if value == self.default_value: - return # Ignore default values - try: - setattr( - instance, - self.private_field_name, - self.serializer.from_dict_value(self.field_type, value), - ) - except Exception as e: - raise + def __init__( + self, + json_name, + field_name, + private_field_name, + field_type, + default_value, + serializer, + optional=False, + ): + self.json_name = json_name + self.field_name = field_name + self.private_field_name = private_field_name + self.field_type = field_type + self.default_value = default_value + self.serializer = serializer + self.optional = optional + + def get_as_dict_item(self, instance, ignore_defaults=True): + value = getattr(instance, self.private_field_name) + if ignore_defaults and value == self.default_value: + return None + if value is None: + return None + return self.serializer.to_dict_value(self.field_type, value, ignore_defaults) + + def set_from_dict(self, instance, json_dict): + if self.json_name not in json_dict: + return # Ignore unknown fields + value = json_dict[self.json_name] + if value == self.default_value: + return # Ignore default values + try: + setattr( + instance, + self.private_field_name, + self.serializer.from_dict_value(self.field_type, value), + ) + except Exception as e: + raise class KaggleObject(object): - def endpoint(self): - raise "Error: endpoint must be defined by the request object" - - @staticmethod - def endpoint_path(): - return None - - @staticmethod - def body_fields(): - return None - - @classmethod - def prepare_from(cls, http_response): - return cls.from_json(http_response.text) - - @staticmethod - def method(): - return "GET" - - def _freeze(self): - self._is_frozen = True - - def __setattr__(self, key, value): - if hasattr(self, "_is_frozen") and not hasattr(self, key): - raise AttributeError(f"Unknown field for {self.__class__.__name__}: {key}") - object.__setattr__(self, key, value) - - def to_dict(self, ignore_defaults=True): - kv_pairs = [(field.json_name, field.get_as_dict_item(self, ignore_defaults)) for field in self._fields] - return {k: v for (k, v) in kv_pairs if not ignore_defaults or v is not None} - - def to_field_map(self, ignore_defaults=True): - kv_pairs = [(field.field_name, field.get_as_dict_item(self, ignore_defaults)) for field in self._fields] - return {k: v for (k, v) in kv_pairs if not ignore_defaults or v is not None} - - @classmethod - def from_dict(cls, json_dict): - instance = cls() - for field in cls._fields: - field.set_from_dict(instance, json_dict) - return instance - - @classmethod - def from_json(cls, json_str): - return cls.from_dict(json.loads(json_str)) - - def to_json(self, ignore_defaults=True): - return json.dumps(self.to_dict(ignore_defaults)) - - def __str__(self): - return self.to_json(ignore_defaults=False) - - def __repr__(self): - return self.to_json(ignore_defaults=False) - - def __contains__(self, field_name): - try: - field = self._get_field(field_name) - except ValueError: - return False - value = getattr(self, field.private_field_name) - if field.optional: - return value is not None - else: - return value != field.default_value - - def __delattr__(self, field_name): - field = self._get_field(field_name) - setattr(self, field.private_field_name, field.default_value) - - def _get_field(self, field_name): - field = next((f for f in self._fields if f.field_name == field_name), None) - if field is None: - raise ValueError(f'Protocol message {self.__class__.__name__} has no "{field_name}" field.') - return field + def endpoint(self): + raise "Error: endpoint must be defined by the request object" + + @staticmethod + def endpoint_path(): + return None + + @staticmethod + def body_fields(): + return None + + @classmethod + def prepare_from(cls, http_response): + return cls.from_json(http_response.text) + + @staticmethod + def method(): + return "GET" + + def _freeze(self): + self._is_frozen = True + + def __setattr__(self, key, value): + if hasattr(self, "_is_frozen") and not hasattr(self, key): + raise AttributeError(f"Unknown field for {self.__class__.__name__}: {key}") + object.__setattr__(self, key, value) + + def to_dict(self, ignore_defaults=True): + kv_pairs = [ + (field.json_name, field.get_as_dict_item(self, ignore_defaults)) + for field in self._fields + ] + return {k: v for (k, v) in kv_pairs if not ignore_defaults or v is not None} + + def to_field_map(self, ignore_defaults=True): + kv_pairs = [ + (field.field_name, field.get_as_dict_item(self, ignore_defaults)) + for field in self._fields + ] + return {k: v for (k, v) in kv_pairs if not ignore_defaults or v is not None} + + @classmethod + def from_dict(cls, json_dict): + instance = cls() + for field in cls._fields: + field.set_from_dict(instance, json_dict) + return instance + + @classmethod + def from_json(cls, json_str): + return cls.from_dict(json.loads(json_str)) + + def to_json(self, ignore_defaults=True): + return json.dumps(self.to_dict(ignore_defaults)) + + def __str__(self): + return self.to_json(ignore_defaults=False) + + def __repr__(self): + return self.to_json(ignore_defaults=False) + + def __contains__(self, field_name): + try: + field = self._get_field(field_name) + except ValueError: + return False + value = getattr(self, field.private_field_name) + if field.optional: + return value is not None + else: + return value != field.default_value + + def __delattr__(self, field_name): + field = self._get_field(field_name) + setattr(self, field.private_field_name, field.default_value) + + def _get_field(self, field_name): + field = next((f for f in self._fields if f.field_name == field_name), None) + if field is None: + raise ValueError( + f'Protocol message {self.__class__.__name__} has no "{field_name}" field.' + ) + return field