Module xmodel.base
Expand source code
from xmodel.base.model import BaseModel
from xmodel.base.api import BaseApi
from xmodel.base.structure import BaseStructure
__all__ = [
"BaseModel",
"BaseStructure",
"BaseApi"
]
Sub-modules
xmodel.base.api
-
If you don't know much about the ORM, read ORM Library Overview first! …
xmodel.base.fields
-
TODO
Write this module doc-comment, here is a link that might be useful; it links back to the the overview doc for field objects at …
xmodel.base.model
xmodel.base.structure
-
See
BaseStructure
for more details; This lets you discover/kee-track/find structural details of aBaseModel.api
.
Classes
class BaseApi (*, api: BaseApi[M] = None, model: BaseModel = None)
-
This class is a sort of "Central Hub" that ties all intrested parties together.
You can get the correct instance via
BaseModel
.In order to reduce any name-collisions for other normal Model attributes, everything related to the BaseApi that the
BaseModel
needs is gotten though via this class.You can get the BaseApi instance related to the model via
BaseModel.api
.Example:
>>> obj = BaseModel.api.get_via_id(1)
For more information see BaseApi Class Overview.
Warning: You can probably skip the rest (below)
Most of the time you don't create
BaseApi
objects your self, and so for most people you can skip the following unless you want to know more about internal details.Init Method Specifics
Normally you would not create an
BaseApi
object directly your self.BaseModel
's know how to do this automatically. It happens inBaseModel.__init_subclass__()
.Details about how the arguments you can pass are below.
BaseModel Class Construction:
If you provide an
xmodel.base.api
arg without axmodel.base.model
arg; we will copy theBaseApi.structure
into new object, resetting the error status, and internalBaseApi._state
to None. Thisxmodel.base.api
object is supposed to be the parent BaseModel's class api object.If both
xmodel.base.api
arg +xmodel.base.model
arg areNone
, the BaseModel is the root/generic BaseModel (ie: it has no parent BaseModel).This is what is done by BaseModel classes while the class is lazily loading and creating/configuring the BaseModel class and it's associated
BaseApi
object (accessible viaBaseModel.api
)BaseModel Instance Creation:
If you also pass in a
xmodel.base.model
arg; this get you a special copy of the api you passed in for use just with that BaseModel instance. The modelBaseApi._state
will be allocated internally in the init'd BaseApi object. This is how aBaseModel
instance get's it's own associatedBaseApi
object (that's a different instance vs the one set on BaseModel class when the BaseModel class was originally constructed).All params are optional.
Args
api
-
The "parent" BaseApi obj to copy the basic structure from as a starting point, etc. The superclasses BaseApi class is passed via this arg. This is only used when allocating a new
BaseApi
object for a newBaseModel
class (not an instance, a model class/type). This BaseApi object is used for the class-level BaseModel api object; ie: via "ModelClass.api"See above "BaseModel Class Construction" for more details.
model
-
BaseModel to associate new BaseApi obj with. This is only used to create a new BaseApi object for a
BaseModel
instance for an already-existing type. ie: for BaseModel object instances.See above "BaseModel Instance Creation" for more details.
Expand source code
class BaseApi(Generic[M]): """ This class is a sort of "Central Hub" that ties all intrested parties together. You can get the correct instance via `xmodel.base.model.BaseModel`. In order to reduce any name-collisions for other normal Model attributes, everything related to the BaseApi that the `xmodel.base.model.BaseModel` needs is gotten though via this class. You can get the BaseApi instance related to the model via `xmodel.base.model.BaseModel.api`. Example: >>> obj = BaseModel.api.get_via_id(1) For more information see [BaseApi Class Overview](#api-class-overview). """ # Defaults Types to use. When you sub-class BaseApi, you can declare/override these type-hints # and specify a different type... The system will allocate and use that new type instead # for you automatically on any instances created of the class. # # The BaseAuth won't modify the request to add auth; so it's safe to use as the base default. # # These are implemented via `@property` methods further below, but these are the type-hints. # # The properties and the __init__ method all use these type-hints in order to use the correct # type for each one on-demand as needed. For details on each one, see the @property method(s). # # PyCharm has some sort of issue, if I provide property type-hint and then a property function # that implements it. For some reason, this makes it ignore the type-hint in subclasses # but NOT in the current class. It's some sort of bug. I get around this bug by using a # different initial-name for the property by pre-pending a `_` to it, # and then setting it to the correct name later. # # Example: `structure = _structure` is done after `def _structure(...)` is defined. # See `_structure` method for docs and method that gets this value. structure: BaseStructure[Field] @property def _structure(self): """ Contain things that don't vary among the model instances; ie: This is the same object and applies to all instances of a particular BaseModel class. This object has a list of `xmodel.fields.Field` that apply to the `xmodel.base.model.BaseModel` you can get via `xmodel.base.structure.Structure.fields`; for example. This is currently created in `BaseApi.__init__`. BaseApi instance for a BaseModel is only created when first asked for via `xmodel.base.model.BaseModel.api`. Returns: BaseStructure: Structure with correct field and model type in it. """ return self._structure # PyCharm has some sort of issue, if I provide property type-hint and then a property function # that implements it. For some reason, this makes it ignore the type-hint in subclasses # but NOT in the current class. It's some sort of bug. This gets around it since pycharm # can't figure out what's going on here. structure = _structure _structure = None """ See `BaseApi.structure`. """ # ------------------------------ # --------- Properties --------- default_converters: Dict[Type[Any], Converter] = None """ For an overview of type-converts, see [Type Converters Overview](./#type-converters). The class attribute defaults to `None`, but an instance/object will always have some sort of dict in place (happens during init call). Notice the `todo` note in the [overview](./#type-converters). I want it to work that way in the future (so via `BaseApi.set_default_converter` and `BaseApi.get_default_converter`). It's something coming in the future. For now you'll need to override `default_converters` and/or change it directly. You can provide your own values for this directly in a sub-class, when an BaseApi or subclass is created, we will merge converters in this order, with things later in the order taking precedence and override it: 1. `xmodel.converters.DEFAULT_CONVERTERS` 2. `BaseApi.default_converters` from `xmodel.base.model.BaseModel.api` from parent model. The parent model is the one the model is directly inheriting from. 3. Finally, `BaseApi.default_converters` from the BaseApi subclass's class attribute (only looks on type/class directly for `default_converters`). It takes this final mapping and sets it on `self.default_converters`, and will be inherited as explained on on line number `2` above in the future. Default converters we have defined at the moment: - `xmodel.converters.convert_json_date` - `xmodel.converters.convert_json_datetime` - And a set of basic converters via `xmodel.converters.ConvertBasicType`, supports: - float - bool - str - int See `xmodel.converters.DEFAULT_CONVERTERS` to see the default converters map/dict. Maps type-hint to a default converter. This converter will be used for `TypeValue.convert` when the model BaseStructure is create if none is provided for it at field definition time for a particular type-hint. If a type-hint is not in this converter, no convert is called for it. You don't need to provide one of these for a `xmodel.base.model.BaseModel` type-hint, as the system knows to call json/update_from_json on those types of objects. The default value provides a way to convert to/from a dt.date/dt.datetime and a string. """ # def set_default_converter(self, type, converter): # """ NOT IMPLEMENTED YET - # .. Todo:: Josh: These were here to look up a converter from a parent if a child does not # have one I have not figured out what I want to do here quite yet... # # See todo at [Type Converters](./#type-converters) for an explanation of what this may # be in the future. # # """ # raise NotImplementedError() # # def get_default_converter(self, type) -> Optional[Converter]: # """ NOT IMPLEMENTED YET - # .. Todo:: Josh: These were here to look up a converter from a parent if a child does not # have one I have not figured out what I want to do here quite yet... # # See todo at [Type Converters](./#type-converters) for an explanation of what this may # be in the future. # """ # raise NotImplementedError() # ------------------------------ # --------- Properties --------- @property def model_type(self) -> Type[M]: """ The same BaseApi class is meant to be re-used for any number of Models, and so a BaseModel specifies it's BaseApi type as generic `BaseApi[M]`. In this case is the BaseModel it's self. That way we can have the type-system aware that different instances of the same BaseApi class can specify different associated BaseModel classes. This property will return the BaseModel type/class associated with this BaseApi instance. """ # noinspection PyTypeChecker return self.structure.model_cls # used as an internal class property _CACHED_TYPE_HINTS = {} @classmethod def resolved_type_hints(cls) -> Dict[str, Type]: if hints := BaseApi._CACHED_TYPE_HINTS.get(cls): return hints hints = get_type_hints(cls) BaseApi._CACHED_TYPE_HINTS[cls] = hints return hints # --------------------------- # --------- Methods --------- # noinspection PyMissingConstructor def __init__(self, *, api: "BaseApi[M]" = None, model: BaseModel = None): """ .. warning:: You can probably skip the rest (below) Most of the time you don't create `BaseApi` objects your self, and so for most people you can skip the following unless you want to know more about internal details. # Init Method Specifics Normally you would not create an `BaseApi` object directly your self. `xmodel.base.model.BaseModel`'s know how to do this automatically. It happens in `xmodel.base.model.BaseModel.__init_subclass__`. Details about how the arguments you can pass are below. ## BaseModel Class Construction: If you provide an `api` arg without a `model` arg; we will copy the `BaseApi.structure` into new object, resetting the error status, and internal `BaseApi._state` to None. This `api` object is supposed to be the parent BaseModel's class api object. If both `api` arg + `model` arg are `None`, the BaseModel is the root/generic BaseModel (ie: it has no parent BaseModel). This is what is done by BaseModel classes while the class is lazily loading and creating/configuring the BaseModel class and it's associated `BaseApi` object (accessible via `xmodel.base.model.BaseModel.api`) ## BaseModel Instance Creation: If you also pass in a `model` arg; this get you a special copy of the api you passed in for use just with that BaseModel instance. The model `BaseApi._state` will be allocated internally in the init'd BaseApi object. This is how a `xmodel.base.model.BaseModel` instance get's it's own associated `BaseApi` object (that's a different instance vs the one set on BaseModel class when the BaseModel class was originally constructed). All params are optional. Args: api: The "parent" BaseApi obj to copy the basic structure from as a starting point, etc. The superclasses BaseApi class is passed via this arg. This is only used when allocating a new `BaseApi` object for a new `xmodel.base.model.BaseModel` class (not an instance, a model class/type). This BaseApi object is used for the class-level BaseModel api object; ie: via "ModelClass.api" See above "BaseModel Class Construction" for more details. model: BaseModel to associate new BaseApi obj with. This is only used to create a new BaseApi object for a `xmodel.base.model.BaseModel` instance for an already-existing type. ie: for BaseModel object instances. See above "BaseModel Instance Creation" for more details. """ if api and model: raise XModelError( f"You can't pass in an BaseApi {api} and BaseModel {model} simultaneously." ) if model: api = type(model).api if not api: assert not model, "You can't pass in a model without an associated api/model obj." if model: # If we have a model, the structure should be exactly the same as it's BaseModel type. self._structure = api.structure self._api_state = PrivateApiState(model=model) self.default_converters = api.default_converters return # If We don't have a BaseModel, then we need to copy the structure, it could change # because we are being allocated for a new BaseModel type at the class/type level; # this means we are not associated with a specific BaseModel instance, only a BaseModel # type. # We lookup the structure type that our associated model-type/class wants to use. structure_type = type(self).resolved_type_hints().get( 'structure', BaseStructure[Field] ) args = typing_inspect.get_args(structure_type) field_type = args[0] if args else Field # We have a root BaseModel with the abstract BaseModel as its super class, # in this case we need to allocate a blank structure object. # todo: allocate structure with new args # We look up the structure type that our associated model-type/class wants to use. existing_struct = api.structure if api else None self._structure = structure_type( parent=existing_struct, field_type=field_type ) # default_converters is a mapping of type to convert too, and a converter callable. # # We want to inherit from the parent and converters they already have defined. # # Take any parent converters as they currently exist, and use them as a basis for our # converters. Then take any converters directly assigned to self and override the any # parent converters, when they both have a converter for the same key/type. self.default_converters = { **DEFAULT_CONVERTERS, **(api.default_converters or {} if api else {}), **(type(self).default_converters or {}), } # ---------------------------------------------------- # --------- Things REQUIRING an Associated BaseModel ----- @property def model(self) -> M: """ REQUIRES associated model object [see doc text below]. Gives you back the model associated with this api. If this BaseApi obj is associated directly with the BaseModel class type and so there is no associated model, I will raise an exception. Some BaseApi methods are dependant on having an associated model, and when they ask for it and there is None, this will raise an exception for them. The first line of the doc comment tells you if it needs one. Normally, it's pretty obvious if the method will need the model, due to what it will return to you (ie: if it would need model attrs). The methods that are dependant on a model are ones, like 'json', where it returns the JSON for a model. It needs a model to get this data. If you access an object api via a BaseModel object, that will be the associated model. If you access it via a BaseModel type/class, it will be directly associated with the model class. Examples: >>> # Grab Account model from some_lib (as an example). >>> from some_lib.account import Account >>> >>> # api object is associated with MyModelClass class, not model obj. >>> Account.api >>> >>> account_obj = Account.api.get_via_id(3) >>> # api is associated with the account_obj model object. >>> account_obj.api >>> >>> # This sends object attributes to API, so it needs an associated >>> # BaseModel object, so this works: >>> account_obj.api.send() >>> >>> # This would produce an exception, since it would try to get BaseModel >>> # attributes to send. But there is no associated model. >>> Account.api.send() """ api_state = self._api_state assert api_state, "BaseApi needs an attached model obj and there is no associated " \ "model api state." model = api_state.model assert model, "BaseApi needs an attached model obj and there is none." return model def get_child_without_lazy_lookup( self, child_field_name, *, false_if_not_set=False, ) -> Union[M, None, bool, NullType]: """ REQUIRES associated model object [see self.model]. If the child is current set to Null, or an object, returns that value. Will NOT lazily lookup child, even if its possible to do so. :param child_field_name: The field name of the child object. :param false_if_not_set: Possible Values [Default: False]: * False: Return None if nothing is currently set. * True: Return False if nothing is currently set. This lets you distinguish between having a None value set on field vs nothing set at all. Normally this distinction is only useful internally in this class, external users probably don't need this option. """ model = self.model if not self.structure.is_field_a_child(child_field_name): raise XModelError( f"Called get_child_without_lazy_lookup('{child_field_name}') but " f"field ({child_field_name}) is NOT a child field on model ({model}).") if child_field_name in model.__dict__: return getattr(model, child_field_name) if false_if_not_set: return False return None @property def have_changes(self) -> bool: """ Is True if `self.json(only_include_changes=True)` is not None; see json() method for more details. """ log.debug(f"Checking Obj {self.model} to see if I have any changes [have_changes]") return self.json(only_include_changes=True) is not None def json( self, only_include_changes: bool = False, log_output: bool = False, include_removals: bool = False ) -> Optional[JsonDict]: """ REQUIRES associated model object (see `BaseApi.model` for details on this). Return associated model object as a JsonDict (str keys, any value), ready to be encoded via JSON encoder and sent to the API. Args: only_include_changes: If True, will only include what changed in the JsonDict result. Defaults to False. This is normally set to True if system is sending this object via PATCH, which is the normal way the system sends objects to API. If only_include_changes is False (default), we always include everything that is not 'None'. When a `xmodel.base.client.BaseClient` subclass (such as `xmodel.rest.RestClient`) calls this method, it will pass in a value based on it's own `xmodel.rest.RestClient.enable_send_changes_only` is set to (defaults to False there too). You can override the RestClient.enable_send_changes_only at the BaseModel class level by making a RestClient subclass and setting `enable_send_changes_only` to default to `True`. There is a situations where we have to include all attributes, regardless: 1. If the 'id' field is set to a 'None' value. This indicates we need to create a new object, and we are not partially updating an existing one, even if we got updated via json at some point in the past. As always, properties set to None will *NOT* be included in returned JsonDict, regardless of what options have been set. log_output (bool): If False (default): won't log anything. If True: Logs what method returns at debug level. include_removals (bool): If False (default): won't include in response any fields that have been removed (vs when compared to the original JSON that updated this object). The value will be the special sentinel object `Remove` (see top of this module/file for `Remove` object, and it's `RemoveType` class). Returns: JsonDict: Will the needed attributes that should be sent to API. If returned value is None, that means only_include_changes is True and there were no changes. The returned dict is a copy and so can be mutated be the caller. """ # todo: Refactor _get_fields() to return getter/setter closures for each field, and we # can make this whole method more generic that way. We also can 'cache' the logic # needed that way instead of having to figure it out each time, every time. structure = self.structure model = self.model api_state = self._api_state json: JsonDict = {} field_objs = structure.fields # Negate only_include_changes if we don't have any original update json to compare against. if only_include_changes and api_state.last_original_update_json is None: only_include_changes = False # noinspection PyDefaultArgument def set_value_into_json_dict(value, field_name, *, json=json): # Sets field value directly on json dict or passed in dict... if value is not None: # Convert Null into None (that's how JSON converter represents a Null). json[field_name] = value if value is not Null else None for field_obj in field_objs: # If we are read-only, no need to do anything more. if field_obj.read_only: continue # We deal with non-related types later. related_type = field_obj.related_type if not related_type: continue f = field_obj.name if field_obj.read_only: continue # todo: For now, the 'api-field-path' option can't be used at the same time as obj-r. if field_obj.json_path != field_obj.name: # I've put in some initial support for this below, but it's has not been tested # for now, keep raising an exception for this like we have been. # There is a work-around, see bottom part of the message in the below error: raise NotImplementedError( f"Can't have xmodel.Field on BaseModel with related-type and a json_path " f"that differ at the moment, for field ({field_obj}). " f"It is something I want to support someday; the support is mostly in place " f"already, but it needs some more careful thought, attention and testing " f"before we should allow it. " "Workaround: Make an `{field.name}_id` field next to related field on the " "model. Then, set `json_path` for that `{field.name}_id` field, set it to " "what you want it to be. Finally, set the `{related_field.name}` to " "read_only=True. This allows you to rename the `_id` field used to/from api " "in the JSON input/output, but the Model can have an alternate name for the " "related field. You can see a real-example of this at " "`bigcommerce.api.orders._BcCommonOrderMetafield.order" ) obj_type_structure = related_type.api.structure obj_type_has_id = obj_type_structure.has_id_field() if obj_type_has_id: # If the obj uses an 'id', then we have a {field_name}_id we want to # send instead of the full object as a json dict. # # This will grab the id from child obj if it exists, or from a defined field # of f"{f}_id" or finally from related id storage. # todo: If there is an object with no 'id' value, do we ignore it? # or should we embed full object anyway? child_obj_id = api_state.get_related_field_id(f) # Method below should deal with None vs Null. set_value_into_json_dict(child_obj_id, f"{f}_id") else: obj: 'M' = getattr(model, f) # Related-object has no 'id', so get it's json dict and set that into the output. v = obj if obj is not Null and obj is not None: # todo: a Field option to override this and always provide all # values (if object always needs to be fully embedded). v = obj.api.json(only_include_changes=only_include_changes) # if it returns None (ie: no changes) and only_include_changes is enabled, # don't include the sub-object as a change. if v is not None or not only_include_changes: # Method below should deal with None vs Null. set_value_into_json_dict(v, f) for field_obj in field_objs: # If we are read-only, no need to do anything more. if field_obj.read_only: continue # We don't deal with related-types here. if field_obj.related_type: continue f = field_obj.name v = getattr(model, f) if v is not None and field_obj.converter: # Convert the value.... v = field_obj.converter( api=self, direction=Converter.Direction.to_json, field=field_obj, value=v ) path = field_obj.json_path if not path: set_value_into_json_dict(v, f) continue path_list = path.split(field_obj.json_path_separator) d = json for name in path_list[:-1]: d = d.setdefault(name, {}) name = path_list[-1] # Sets field value into a sub-dictionary of the original `json` dict. set_value_into_json_dict(v, name, json=d) if include_removals: removals = self.fields_to_remove_for_json(json, field_objs) for f in removals: if f in json: raise XynModelError( f"Sanity check, we were about to overwrite real value with `Remove` " f"in json field ({f}) for model ({self.model})." ) json[f] = Remove # If the `last_original_update_json` is None, then we never got update via JSON # so there is nothing to compare, include everything! if only_include_changes: log.debug(f"Checking Obj {model} for changes to include.") fields_to_pop = self.fields_to_pop_for_json(json, field_objs, log_output) for f in fields_to_pop: del json[f] if not json: # There were no changes, return None. return None else: due_to_msg = "unknown" if not only_include_changes: due_to_msg = "only_include_changes is False" if api_state.last_original_update_json is None: due_to_msg = "no original json value" if log_output: log.debug(f"Including everything for obj {model} due to {due_to_msg}.") # Log out at debug level what we are including in the JSON. for field, new_value in json.items(): log.debug( f" Included field ({field}) value ({new_value})" ) for k, v in json.items(): # Must use list of JSON, convert any sets to a list. if type(v) is set: v = list(v) json[k] = v return json def fields_to_remove_for_json(self, json: dict, field_objs: List[Field]) -> Set[str]: """ Returns set of fields that should be considered 'changed' because they were removed when compared to the original JSON values used to originally update this object. The names will be the fields json_path. """ fields_to_remove = set() for field in field_objs: # A `None` in the `json` means a null, so we use `Default` as our sentinel type. new_value = json.get(field.json_path, Default) old_value = self._get_old_json_value(field=field.json_path, as_type=type(new_value)) if new_value is Default and old_value is not Default: fields_to_remove.add(field.json_path) return fields_to_remove def fields_to_pop_for_json( self, json: dict, field_objs: List[Field], log_output: bool ) -> Set[Any]: """ Goes through the list of fields (field_objs) to determine which ones have not changed in order to pop them out of the json representation. This method is used when we only want to include the changes in the json. :param json: dict representation of a model's fields and field values as they are currently set on the model. :param field_objs: List of fields and their values for a model :param log_output: boolean to determine if we should log the output or not :return: The field keys to remove from the json representation of the model. """ fields_to_pop = set() for field, new_value in json.items(): # json has simple strings, numbers, lists, dict; # so makes general comparison simpler. old_value = self._get_old_json_value(field=field, as_type=type(new_value)) if old_value is Default: if log_output: log.debug( f" Included field ({field}) with value " f"({new_value}) because there is no original json value for it." ) elif self.should_include_field_in_json( new_value=new_value, old_value=old_value, field=field ): if log_output: log.debug( f" Included field ({field}) due to new value " f"({new_value}) != old value ({old_value})." ) else: # We don't want to mutate dict while traversing it, remember this for later. fields_to_pop.add(field) # Map a field-key to what other fields should be included if field-key value is used. # For now we are NOT supporting `Field.json_path` to keep things simpler # when used in conjunction with `Field.include_with_fields`. # `Field` will raise an exception if json_path != field name and include_with_fields # is used at the same time. # It's something I would like to support in the future, but for now it's not needed. # We can assume that `field_obj.name == field_obj.json_path` for field_obj in field_objs: if not field_obj.include_with_fields: continue if field_obj.name not in fields_to_pop: continue if not (field_obj.include_with_fields <= fields_to_pop): fields_to_pop.remove(field_obj.name) return fields_to_pop def should_include_field_in_json(self, new_value: Any, old_value: Any, field: str) -> bool: """ Returns True if the value for field should be included in the JSON. This only gets called if only_include_changes is True when passed to self.json:: # Passed in like so: self.json(only_include_changes=True) This method is an easy way to change the comparison logic. `new_value` could be `xyn_types.default.Default`, to indicate that a value's absence is significant (ie: to remove an attribute from destination). Most of the time, a value's absence does not affect the destination when object is sent to API/Service because whatever value is currently there for the attribute is left intact/alone. But sometimes a service will remove the attribute if it does not exist. When this is the case, the absence of the value is significant for comparison purposes; ie: when deciding if a value has changed. :param new_value: New value that will be put into JSON. :param old_value: Old value originals in original JSON [normalized if possible to the same type as new_value. :param field: Field name. :return: If True: Will include the fields value in an update. If False: Won't include the fields value in an update. """ # Convert old value to set if new value is set and old value is list (from original JSON). # If I was really cool :)... I would find out the inner type in case of int/str # and to a conversion to compare Apples to Apples..... # But trying to minimize changes so I don't conflict as much with soon to be # xdynamo feature. if type(new_value) is set and type(old_value) is list: old_value = set(old_value) return new_value != old_value def _get_old_json_value(self, *, field: str, as_type: Type = None) -> Optional[Any]: """ Returns the old field-values; Will return `Default` if there is no original value. """ original_json = self._api_state.last_original_update_json if original_json is None: # todo: Is there another value we could return here to indicate that we # never got an original value in the first place? # # todo: Also, think about how we could do above todo ^ per-field # [ie: if field was requested in the first place]. return Default old_value = original_json.get(field, Default) if old_value is Default: # None is a valid value in JSON, # this indicates to do the Default thing/value with this field since we don't have any # original value for it. return Default # json has simple strings, numbers, lists, dict; # so makes general comparison simpler. old_type = type(old_value) if as_type != old_type: str_compatible_types = {str, int, float} if as_type in str_compatible_types and old_type in str_compatible_types: try: # The 'id' field is a string and not an int [for example], so in # general, we want to try and convert the old value into the new # values type before comparison, if possible, for the basic types # of str, int, float. old_value = as_type(old_value) except ValueError: # Just be sure it's the same value/type, should be but just in case. old_value = original_json.get(field, None) pass return old_value def copy_from_model(self, model: BaseModel): their_fields = model.api.structure.field_map my_fields = self.structure.field_map keys = [k for k in their_fields if k in my_fields] # Assume we have a model, and are not the class-based `MyModel.api....` version. # todo: have `self.model` raise an exception if called on the class api version # (which does not have a related model, just knows about model-type.). my_model = self.model for k in keys: their_value = getattr(model, k) if their_value is not None: setattr(my_model, k, their_value) def update_from_json(self, json: Union[JsonDict, Mapping]): """ REQUIRES associated model object [see self.model]. todo: Needs more documentation We update the dict per-key, with what we got passed in [via 'json' parameter] overriding anything we got previously. This also makes a copy of the dict, which is want we want [no change to modify the incoming dict parameter]. """ structure = self.structure model = self.model api_state = self._api_state if not isinstance(json, Mapping): raise XModelError( f"update_from_json(...) was given a non-mapping parameter ({json})." ) # Merge the old values with the new values. api_state.last_original_update_json = { **(api_state.last_original_update_json or {}), **json } fields = structure.fields values = {} for field_obj in fields: path_list = field_obj.json_path.split(field_obj.json_path_separator) v = json got_value = True for name in path_list: if name not in v: # We don't even have a 'None' value so we assume we just did not get the value # from the api, and therefore we just skip doing anything with it. got_value = False break v = v.get(name) if v is None: break # We map the value we got from JSON into a flat-dict with the BaseModel name as the # key... if got_value: values[field_obj.name] = v if v is not None else Null def set_attr_on_model(field, value, model=model): """ Closure to set attr on self unless value is None. """ if value is None: return setattr(model, field, value) # Merge in the outer json, keeping the values we mapped [via Field.json_path] for conflicts values = {**json, **values} # todo: If the json does not have a value [not even a 'None' value], don't update? # We may have gotten a partial update? For now, always update [even to None] # all defined fields regardless if they are inside the json or not. for field_obj in fields: # We deal with related types later.... if field_obj.related_type: continue f = field_obj.name v = values.get(f, Default) # A None from JSON means a Null for us. # If JSON does not include anything, that's a None for us. if v is None: v = Null elif v is Default: v = None # Run the converter if needed. # If we have a None value, we don't need to convert that, there was no value to # convert. if field_obj.converter and v is not None: v = field_obj.converter( self, Converter.Direction.from_json, field_obj, v ) set_attr_on_model(f, v) for field_obj in fields: # Ok, now we deal with related types... related_type = field_obj.related_type if not related_type: continue f = field_obj.name # todo: at some point, allow customization of this via Field class # Also, s tore the id f_id_name = f"{f}_id" if typing_inspect.get_origin(field_obj.type_hint) is list: # todo: This code is not complete [Kaden never finished it up] # for now, just comment out. raise NotImplementedError( "Type-hints for xmodel models in this format: `attr: List[SomeType]` " "are not currently supported. We want to support it someday. For now you " "must use lower-cased non-generic `list`. At some point the idea is to " "allow one to do `List[ChildModel]` and then we know it's a list of " "other BaseModel objects and automatically handle that in some way." ) # child_type: 'Type[M]' # child_type = typing_inspect.get_args(obj_type) # # __args__ returns a tuple of all arguments passed into List[] so we need to # # pull the class out of the tuple # if child_type: # child_type = child_type[0] # # child_api: BaseApi # child_api = child_type.api # if not child_api and child_api.structure.has_id_field: # # TODO: add a non generic Exception for this # raise XModelError( # f"{model} has an attribute with name ({f}) with type-hint List that " # f"doesn't contain an API BaseModel Type as the only argument" # ) # parent_name = model.__class__.__name__.lower() # state.set_related_field_id(f, parent_name) # continue v = None if f in values: v = values.get(f, Null) if v is not Null: v = related_type(v) # Check to see if we have an api/json field for object relation name with "_id" on # end. if v is None and related_type.api.structure.has_id_field(): # If we don't have a defined field for this value, check JSON for it and store it. # # If we have a defined None value for the id field, meaning the field exists # in the json, and is set directly to None, then we have a Null relationship. # We set that as the value, since there is no need to 'lookup' a null value. f_id_value = json.get(f_id_name) id_field = structure.get_field(f_id_name) if not id_field: id_field = field_obj.related_type.api.structure.get_field('id') # Run the converter if needed. # If we have a None value, we don't need to convert that, there was no value to # convert. if id_field and id_field.converter and f_id_value is not None: f_id_value = id_field.converter( self, Converter.Direction.from_json, id_field, f_id_value ) if f_id_value is None and f_id_name in json: # We have a Null situation. f_id_value = Null if f_id_value is not None: # We have an id! # Set the value to support automatic lookup of value, lazily. # This method also takes care to set child object to Null or delete it # as needed depending on the f_id_value and what the child's id field value is. api_state.set_related_field_id(f, f_id_value) else: # 'v' is either going to be None, Null or an BaseModel object. set_attr_on_model(f, v) def list_of_attrs_to_repr(self) -> List[str]: """" REQUIRES associated model object [see self.model]. A list of attribute names to put into the __repr__/string representation of the associated model object. This is consulted when the BaseModel has __repr__ called on it. """ names = set() model = self.model # todo: Move this into pres-club override of list_of_attrs_to_repr in an BaseApi subclass. if hasattr(model, 'account_id'): names.add('account_id') # todo: Consider adding others here, perhaps all defined fields on model that have # todo: a non-None value? for f in self.structure.fields: if f.include_in_repr: names.add(f.name) return list(names) def forget_original_json_state(self): """ If called, we forget/reset the orginal json state, which is a combination of all the json that this object has been updated with over it's lifetime. The json state is what allows the object to decide what has changed, when it's requested to only include changes via the `BaseApi.json` method. If forgotten, it's as-if we never got the json in the first place to compare against. Therefore, all attributes that have values will be returned for this object when it's only requested to include changes (the RestClient in xmodel-rest can request it to do this, as an example). Resetting the state here only effects this object, not any child objects. You'll have to ask child objects directly to forget t heir original json, if desired. """ self._api_state.last_original_update_json = None # ---------------------------- # --------- Private ---------- # # I want to make the state and structure private for now, because it might change a bit later. # Want to give this some opportunity to be used for a while to see where the areas for # improvement are before potentially opening it up publicly to things outside of the sdk. _api_state: PrivateApiState[M] = None """ This object will vary from BaseModel class instance-to-instance, and is the area we keep api state that is Private for the BaseModel instance. Will be None if we are directly associated with BaseModel class, otherwise this will be the BaseModel's instance state, methods in this object need the BaseModel instance. """ @property def context(self) -> XContext: """ BaseApi context to use when asking this object to send/delete/etc its self to/from service. This is an old hold-over from when we used to keep a XContext reference. This is the same as calling `xinject.context.XContext.current`. """ return XContext.grab()
Ancestors
- typing.Generic
Subclasses
Class variables
var default_converters : Dict[Type[Any], Converter]
-
For an overview of type-converts, see Type Converters Overview.
The class attribute defaults to
None
, but an instance/object will always have some sort of dict in place (happens during init call).Notice the
todo
note in the overview. I want it to work that way in the future (so viaBaseApi.set_default_converter
andBaseApi.get_default_converter
). It's something coming in the future. For now you'll need to overridedefault_converters
and/or change it directly.You can provide your own values for this directly in a sub-class, when an BaseApi or subclass is created, we will merge converters in this order, with things later in the order taking precedence and override it:
xmodel.converters.DEFAULT_CONVERTERS
BaseApi.default_converters
fromBaseModel.api
from parent model. The parent model is the one the model is directly inheriting from.- Finally,
BaseApi.default_converters
from the BaseApi subclass's class attribute (only looks on type/class directly fordefault_converters
).
It takes this final mapping and sets it on
self.default_converters
, and will be inherited as explained on on line number2
above in the future.Default converters we have defined at the moment:
convert_json_date()
convert_json_datetime()
- And a set of basic converters via
ConvertBasicType
, supports:- float
- bool
- str
- int
See
xmodel.converters.DEFAULT_CONVERTERS
to see the default converters map/dict.Maps type-hint to a default converter. This converter will be used for
TypeValue.convert
when the model BaseStructure is create if none is provided for it at field definition time for a particular type-hint. If a type-hint is not in this converter, no convert is called for it.You don't need to provide one of these for a
BaseModel
type-hint, as the system knows to call json/update_from_json on those types of objects.The default value provides a way to convert to/from a dt.date/dt.datetime and a string.
Static methods
def resolved_type_hints() ‑> Dict[str, Type]
-
Expand source code
@classmethod def resolved_type_hints(cls) -> Dict[str, Type]: if hints := BaseApi._CACHED_TYPE_HINTS.get(cls): return hints hints = get_type_hints(cls) BaseApi._CACHED_TYPE_HINTS[cls] = hints return hints
Instance variables
var context : XContext
-
BaseApi context to use when asking this object to send/delete/etc its self to/from service.
This is an old hold-over from when we used to keep a XContext reference. This is the same as calling
XContext.current()
.Expand source code
@property def context(self) -> XContext: """ BaseApi context to use when asking this object to send/delete/etc its self to/from service. This is an old hold-over from when we used to keep a XContext reference. This is the same as calling `xinject.context.XContext.current`. """ return XContext.grab()
var have_changes : bool
-
Is True if
self.json(only_include_changes=True)
is not None; see json() method for more details.Expand source code
@property def have_changes(self) -> bool: """ Is True if `self.json(only_include_changes=True)` is not None; see json() method for more details. """ log.debug(f"Checking Obj {self.model} to see if I have any changes [have_changes]") return self.json(only_include_changes=True) is not None
var model : ~M
-
REQUIRES associated model object [see doc text below].
Gives you back the model associated with this api. If this BaseApi obj is associated directly with the BaseModel class type and so there is no associated model, I will raise an exception.
Some BaseApi methods are dependant on having an associated model, and when they ask for it and there is None, this will raise an exception for them. The first line of the doc comment tells you if it needs one. Normally, it's pretty obvious if the method will need the model, due to what it will return to you (ie: if it would need model attrs).
The methods that are dependant on a model are ones, like 'json', where it returns the JSON for a model. It needs a model to get this data.
If you access an object api via a BaseModel object, that will be the associated model. If you access it via a BaseModel type/class, it will be directly associated with the model class.
Examples:
>>> # Grab Account model from some_lib (as an example). >>> from some_lib.account import Account >>> >>> # api object is associated with MyModelClass class, not model obj. >>> Account.api >>> >>> account_obj = Account.api.get_via_id(3) >>> # api is associated with the account_obj model object. >>> account_obj.api >>> >>> # This sends object attributes to API, so it needs an associated >>> # BaseModel object, so this works: >>> account_obj.api.send() >>> >>> # This would produce an exception, since it would try to get BaseModel >>> # attributes to send. But there is no associated model. >>> Account.api.send()
Expand source code
@property def model(self) -> M: """ REQUIRES associated model object [see doc text below]. Gives you back the model associated with this api. If this BaseApi obj is associated directly with the BaseModel class type and so there is no associated model, I will raise an exception. Some BaseApi methods are dependant on having an associated model, and when they ask for it and there is None, this will raise an exception for them. The first line of the doc comment tells you if it needs one. Normally, it's pretty obvious if the method will need the model, due to what it will return to you (ie: if it would need model attrs). The methods that are dependant on a model are ones, like 'json', where it returns the JSON for a model. It needs a model to get this data. If you access an object api via a BaseModel object, that will be the associated model. If you access it via a BaseModel type/class, it will be directly associated with the model class. Examples: >>> # Grab Account model from some_lib (as an example). >>> from some_lib.account import Account >>> >>> # api object is associated with MyModelClass class, not model obj. >>> Account.api >>> >>> account_obj = Account.api.get_via_id(3) >>> # api is associated with the account_obj model object. >>> account_obj.api >>> >>> # This sends object attributes to API, so it needs an associated >>> # BaseModel object, so this works: >>> account_obj.api.send() >>> >>> # This would produce an exception, since it would try to get BaseModel >>> # attributes to send. But there is no associated model. >>> Account.api.send() """ api_state = self._api_state assert api_state, "BaseApi needs an attached model obj and there is no associated " \ "model api state." model = api_state.model assert model, "BaseApi needs an attached model obj and there is none." return model
var model_type : Type[~M]
-
The same BaseApi class is meant to be re-used for any number of Models, and so a BaseModel specifies it's BaseApi type as generic
BaseApi[M]
. In this case is the BaseModel it's self. That way we can have the type-system aware that different instances of the same BaseApi class can specify different associated BaseModel classes.This property will return the BaseModel type/class associated with this BaseApi instance.
Expand source code
@property def model_type(self) -> Type[M]: """ The same BaseApi class is meant to be re-used for any number of Models, and so a BaseModel specifies it's BaseApi type as generic `BaseApi[M]`. In this case is the BaseModel it's self. That way we can have the type-system aware that different instances of the same BaseApi class can specify different associated BaseModel classes. This property will return the BaseModel type/class associated with this BaseApi instance. """ # noinspection PyTypeChecker return self.structure.model_cls
var structure : BaseStructure[Field]
-
Contain things that don't vary among the model instances; ie: This is the same object and applies to all instances of a particular BaseModel class.
This object has a list of
xmodel.fields.Field
that apply to theBaseModel
you can get viaxmodel.base.structure.Structure.fields
; for example.This is currently created in
BaseApi
.BaseApi instance for a BaseModel is only created when first asked for via
BaseModel.api
.Returns
BaseStructure
- Structure with correct field and model type in it.
Expand source code
@property def _structure(self): """ Contain things that don't vary among the model instances; ie: This is the same object and applies to all instances of a particular BaseModel class. This object has a list of `xmodel.fields.Field` that apply to the `xmodel.base.model.BaseModel` you can get via `xmodel.base.structure.Structure.fields`; for example. This is currently created in `BaseApi.__init__`. BaseApi instance for a BaseModel is only created when first asked for via `xmodel.base.model.BaseModel.api`. Returns: BaseStructure: Structure with correct field and model type in it. """ return self._structure
Methods
def copy_from_model(self, model: BaseModel)
-
Expand source code
def copy_from_model(self, model: BaseModel): their_fields = model.api.structure.field_map my_fields = self.structure.field_map keys = [k for k in their_fields if k in my_fields] # Assume we have a model, and are not the class-based `MyModel.api....` version. # todo: have `self.model` raise an exception if called on the class api version # (which does not have a related model, just knows about model-type.). my_model = self.model for k in keys: their_value = getattr(model, k) if their_value is not None: setattr(my_model, k, their_value)
def fields_to_pop_for_json(self, json: dict, field_objs: List[Field], log_output: bool) ‑> Set[Any]
-
Goes through the list of fields (field_objs) to determine which ones have not changed in order to pop them out of the json representation. This method is used when we only want to include the changes in the json.
:param json: dict representation of a model's fields and field values as they are currently set on the model. :param field_objs: List of fields and their values for a model :param log_output: boolean to determine if we should log the output or not :return: The field keys to remove from the json representation of the model.
Expand source code
def fields_to_pop_for_json( self, json: dict, field_objs: List[Field], log_output: bool ) -> Set[Any]: """ Goes through the list of fields (field_objs) to determine which ones have not changed in order to pop them out of the json representation. This method is used when we only want to include the changes in the json. :param json: dict representation of a model's fields and field values as they are currently set on the model. :param field_objs: List of fields and their values for a model :param log_output: boolean to determine if we should log the output or not :return: The field keys to remove from the json representation of the model. """ fields_to_pop = set() for field, new_value in json.items(): # json has simple strings, numbers, lists, dict; # so makes general comparison simpler. old_value = self._get_old_json_value(field=field, as_type=type(new_value)) if old_value is Default: if log_output: log.debug( f" Included field ({field}) with value " f"({new_value}) because there is no original json value for it." ) elif self.should_include_field_in_json( new_value=new_value, old_value=old_value, field=field ): if log_output: log.debug( f" Included field ({field}) due to new value " f"({new_value}) != old value ({old_value})." ) else: # We don't want to mutate dict while traversing it, remember this for later. fields_to_pop.add(field) # Map a field-key to what other fields should be included if field-key value is used. # For now we are NOT supporting `Field.json_path` to keep things simpler # when used in conjunction with `Field.include_with_fields`. # `Field` will raise an exception if json_path != field name and include_with_fields # is used at the same time. # It's something I would like to support in the future, but for now it's not needed. # We can assume that `field_obj.name == field_obj.json_path` for field_obj in field_objs: if not field_obj.include_with_fields: continue if field_obj.name not in fields_to_pop: continue if not (field_obj.include_with_fields <= fields_to_pop): fields_to_pop.remove(field_obj.name) return fields_to_pop
def fields_to_remove_for_json(self, json: dict, field_objs: List[Field]) ‑> Set[str]
-
Returns set of fields that should be considered 'changed' because they were removed when compared to the original JSON values used to originally update this object.
The names will be the fields json_path.
Expand source code
def fields_to_remove_for_json(self, json: dict, field_objs: List[Field]) -> Set[str]: """ Returns set of fields that should be considered 'changed' because they were removed when compared to the original JSON values used to originally update this object. The names will be the fields json_path. """ fields_to_remove = set() for field in field_objs: # A `None` in the `json` means a null, so we use `Default` as our sentinel type. new_value = json.get(field.json_path, Default) old_value = self._get_old_json_value(field=field.json_path, as_type=type(new_value)) if new_value is Default and old_value is not Default: fields_to_remove.add(field.json_path) return fields_to_remove
def forget_original_json_state(self)
-
If called, we forget/reset the orginal json state, which is a combination of all the json that this object has been updated with over it's lifetime.
The json state is what allows the object to decide what has changed, when it's requested to only include changes via the
BaseApi.json()
method.If forgotten, it's as-if we never got the json in the first place to compare against. Therefore, all attributes that have values will be returned for this object when it's only requested to include changes (the RestClient in xmodel-rest can request it to do this, as an example).
Resetting the state here only effects this object, not any child objects. You'll have to ask child objects directly to forget t heir original json, if desired.
Expand source code
def forget_original_json_state(self): """ If called, we forget/reset the orginal json state, which is a combination of all the json that this object has been updated with over it's lifetime. The json state is what allows the object to decide what has changed, when it's requested to only include changes via the `BaseApi.json` method. If forgotten, it's as-if we never got the json in the first place to compare against. Therefore, all attributes that have values will be returned for this object when it's only requested to include changes (the RestClient in xmodel-rest can request it to do this, as an example). Resetting the state here only effects this object, not any child objects. You'll have to ask child objects directly to forget t heir original json, if desired. """ self._api_state.last_original_update_json = None
def get_child_without_lazy_lookup(self, child_field_name, *, false_if_not_set=False) ‑> Union[~M, ForwardRef(None), bool, NullType]
-
REQUIRES associated model object [see self.model].
If the child is current set to Null, or an object, returns that value. Will NOT lazily lookup child, even if its possible to do so.
:param child_field_name: The field name of the child object. :param false_if_not_set: Possible Values [Default: False]: * False: Return None if nothing is currently set. * True: Return False if nothing is currently set. This lets you distinguish between having a None value set on field vs nothing set at all. Normally this distinction is only useful internally in this class, external users probably don't need this option.
Expand source code
def get_child_without_lazy_lookup( self, child_field_name, *, false_if_not_set=False, ) -> Union[M, None, bool, NullType]: """ REQUIRES associated model object [see self.model]. If the child is current set to Null, or an object, returns that value. Will NOT lazily lookup child, even if its possible to do so. :param child_field_name: The field name of the child object. :param false_if_not_set: Possible Values [Default: False]: * False: Return None if nothing is currently set. * True: Return False if nothing is currently set. This lets you distinguish between having a None value set on field vs nothing set at all. Normally this distinction is only useful internally in this class, external users probably don't need this option. """ model = self.model if not self.structure.is_field_a_child(child_field_name): raise XModelError( f"Called get_child_without_lazy_lookup('{child_field_name}') but " f"field ({child_field_name}) is NOT a child field on model ({model}).") if child_field_name in model.__dict__: return getattr(model, child_field_name) if false_if_not_set: return False return None
def json(self, only_include_changes: bool = False, log_output: bool = False, include_removals: bool = False) ‑> Optional[Dict[str, Any]]
-
REQUIRES associated model object (see
BaseApi.model
for details on this).Return associated model object as a JsonDict (str keys, any value), ready to be encoded via JSON encoder and sent to the API.
Args
only_include_changes: If True, will only include what changed in the JsonDict result. Defaults to False. This is normally set to True if system is sending this object via PATCH, which is the normal way the system sends objects to API.
If only_include_changes is False (default), we always include everything that is not 'None'. When a <code>xmodel.base.client.BaseClient</code> subclass (such as <code>xmodel.rest.RestClient</code>) calls this method, it will pass in a value based on it's own <code>xmodel.rest.RestClient.enable\_send\_changes\_only</code> is set to (defaults to False there too). You can override the RestClient.enable_send_changes_only at the BaseModel class level by making a RestClient subclass and setting <code>enable\_send\_changes\_only</code> to default to <code>True</code>. There is a situations where we have to include all attributes, regardless: 1. If the 'id' field is set to a 'None' value. This indicates we need to create a new object, and we are not partially updating an existing one, even if we got updated via json at some point in the past. As always, properties set to None will *NOT* be included in returned JsonDict, regardless of what options have been set.
log_output (bool): If False (default): won't log anything. If True: Logs what method returns at debug level.
include_removals
:bool
- If False (default): won't include in response any fields
that have been removed
(vs when compared to the original JSON that updated this object).
The value will be the special sentinel object
Remove
(see top of this module/file forRemove
object, and it'sRemoveType
class).
Returns
JsonDict
-
Will the needed attributes that should be sent to API. If returned value is None, that means only_include_changes is True and there were no changes.
The returned dict is a copy and so can be mutated be the caller.
Expand source code
def json( self, only_include_changes: bool = False, log_output: bool = False, include_removals: bool = False ) -> Optional[JsonDict]: """ REQUIRES associated model object (see `BaseApi.model` for details on this). Return associated model object as a JsonDict (str keys, any value), ready to be encoded via JSON encoder and sent to the API. Args: only_include_changes: If True, will only include what changed in the JsonDict result. Defaults to False. This is normally set to True if system is sending this object via PATCH, which is the normal way the system sends objects to API. If only_include_changes is False (default), we always include everything that is not 'None'. When a `xmodel.base.client.BaseClient` subclass (such as `xmodel.rest.RestClient`) calls this method, it will pass in a value based on it's own `xmodel.rest.RestClient.enable_send_changes_only` is set to (defaults to False there too). You can override the RestClient.enable_send_changes_only at the BaseModel class level by making a RestClient subclass and setting `enable_send_changes_only` to default to `True`. There is a situations where we have to include all attributes, regardless: 1. If the 'id' field is set to a 'None' value. This indicates we need to create a new object, and we are not partially updating an existing one, even if we got updated via json at some point in the past. As always, properties set to None will *NOT* be included in returned JsonDict, regardless of what options have been set. log_output (bool): If False (default): won't log anything. If True: Logs what method returns at debug level. include_removals (bool): If False (default): won't include in response any fields that have been removed (vs when compared to the original JSON that updated this object). The value will be the special sentinel object `Remove` (see top of this module/file for `Remove` object, and it's `RemoveType` class). Returns: JsonDict: Will the needed attributes that should be sent to API. If returned value is None, that means only_include_changes is True and there were no changes. The returned dict is a copy and so can be mutated be the caller. """ # todo: Refactor _get_fields() to return getter/setter closures for each field, and we # can make this whole method more generic that way. We also can 'cache' the logic # needed that way instead of having to figure it out each time, every time. structure = self.structure model = self.model api_state = self._api_state json: JsonDict = {} field_objs = structure.fields # Negate only_include_changes if we don't have any original update json to compare against. if only_include_changes and api_state.last_original_update_json is None: only_include_changes = False # noinspection PyDefaultArgument def set_value_into_json_dict(value, field_name, *, json=json): # Sets field value directly on json dict or passed in dict... if value is not None: # Convert Null into None (that's how JSON converter represents a Null). json[field_name] = value if value is not Null else None for field_obj in field_objs: # If we are read-only, no need to do anything more. if field_obj.read_only: continue # We deal with non-related types later. related_type = field_obj.related_type if not related_type: continue f = field_obj.name if field_obj.read_only: continue # todo: For now, the 'api-field-path' option can't be used at the same time as obj-r. if field_obj.json_path != field_obj.name: # I've put in some initial support for this below, but it's has not been tested # for now, keep raising an exception for this like we have been. # There is a work-around, see bottom part of the message in the below error: raise NotImplementedError( f"Can't have xmodel.Field on BaseModel with related-type and a json_path " f"that differ at the moment, for field ({field_obj}). " f"It is something I want to support someday; the support is mostly in place " f"already, but it needs some more careful thought, attention and testing " f"before we should allow it. " "Workaround: Make an `{field.name}_id` field next to related field on the " "model. Then, set `json_path` for that `{field.name}_id` field, set it to " "what you want it to be. Finally, set the `{related_field.name}` to " "read_only=True. This allows you to rename the `_id` field used to/from api " "in the JSON input/output, but the Model can have an alternate name for the " "related field. You can see a real-example of this at " "`bigcommerce.api.orders._BcCommonOrderMetafield.order" ) obj_type_structure = related_type.api.structure obj_type_has_id = obj_type_structure.has_id_field() if obj_type_has_id: # If the obj uses an 'id', then we have a {field_name}_id we want to # send instead of the full object as a json dict. # # This will grab the id from child obj if it exists, or from a defined field # of f"{f}_id" or finally from related id storage. # todo: If there is an object with no 'id' value, do we ignore it? # or should we embed full object anyway? child_obj_id = api_state.get_related_field_id(f) # Method below should deal with None vs Null. set_value_into_json_dict(child_obj_id, f"{f}_id") else: obj: 'M' = getattr(model, f) # Related-object has no 'id', so get it's json dict and set that into the output. v = obj if obj is not Null and obj is not None: # todo: a Field option to override this and always provide all # values (if object always needs to be fully embedded). v = obj.api.json(only_include_changes=only_include_changes) # if it returns None (ie: no changes) and only_include_changes is enabled, # don't include the sub-object as a change. if v is not None or not only_include_changes: # Method below should deal with None vs Null. set_value_into_json_dict(v, f) for field_obj in field_objs: # If we are read-only, no need to do anything more. if field_obj.read_only: continue # We don't deal with related-types here. if field_obj.related_type: continue f = field_obj.name v = getattr(model, f) if v is not None and field_obj.converter: # Convert the value.... v = field_obj.converter( api=self, direction=Converter.Direction.to_json, field=field_obj, value=v ) path = field_obj.json_path if not path: set_value_into_json_dict(v, f) continue path_list = path.split(field_obj.json_path_separator) d = json for name in path_list[:-1]: d = d.setdefault(name, {}) name = path_list[-1] # Sets field value into a sub-dictionary of the original `json` dict. set_value_into_json_dict(v, name, json=d) if include_removals: removals = self.fields_to_remove_for_json(json, field_objs) for f in removals: if f in json: raise XynModelError( f"Sanity check, we were about to overwrite real value with `Remove` " f"in json field ({f}) for model ({self.model})." ) json[f] = Remove # If the `last_original_update_json` is None, then we never got update via JSON # so there is nothing to compare, include everything! if only_include_changes: log.debug(f"Checking Obj {model} for changes to include.") fields_to_pop = self.fields_to_pop_for_json(json, field_objs, log_output) for f in fields_to_pop: del json[f] if not json: # There were no changes, return None. return None else: due_to_msg = "unknown" if not only_include_changes: due_to_msg = "only_include_changes is False" if api_state.last_original_update_json is None: due_to_msg = "no original json value" if log_output: log.debug(f"Including everything for obj {model} due to {due_to_msg}.") # Log out at debug level what we are including in the JSON. for field, new_value in json.items(): log.debug( f" Included field ({field}) value ({new_value})" ) for k, v in json.items(): # Must use list of JSON, convert any sets to a list. if type(v) is set: v = list(v) json[k] = v return json
def list_of_attrs_to_repr(self) ‑> List[str]
-
" REQUIRES associated model object [see self.model].
A list of attribute names to put into the repr/string representation of the associated model object. This is consulted when the BaseModel has repr called on it.
Expand source code
def list_of_attrs_to_repr(self) -> List[str]: """" REQUIRES associated model object [see self.model]. A list of attribute names to put into the __repr__/string representation of the associated model object. This is consulted when the BaseModel has __repr__ called on it. """ names = set() model = self.model # todo: Move this into pres-club override of list_of_attrs_to_repr in an BaseApi subclass. if hasattr(model, 'account_id'): names.add('account_id') # todo: Consider adding others here, perhaps all defined fields on model that have # todo: a non-None value? for f in self.structure.fields: if f.include_in_repr: names.add(f.name) return list(names)
def should_include_field_in_json(self, new_value: Any, old_value: Any, field: str) ‑> bool
-
Returns True if the value for field should be included in the JSON. This only gets called if only_include_changes is True when passed to self.json::
# Passed in like so: self.json(only_include_changes=True)
This method is an easy way to change the comparison logic.
new_value
could bexyn_types.default.Default
, to indicate that a value's absence is significant (ie: to remove an attribute from destination).Most of the time, a value's absence does not affect the destination when object is sent to API/Service because whatever value is currently there for the attribute is left intact/alone.
But sometimes a service will remove the attribute if it does not exist. When this is the case, the absence of the value is significant for comparison purposes; ie: when deciding if a value has changed.
:param new_value: New value that will be put into JSON. :param old_value: Old value originals in original JSON [normalized if possible to the same type as new_value. :param field: Field name. :return: If True: Will include the fields value in an update. If False: Won't include the fields value in an update.
Expand source code
def should_include_field_in_json(self, new_value: Any, old_value: Any, field: str) -> bool: """ Returns True if the value for field should be included in the JSON. This only gets called if only_include_changes is True when passed to self.json:: # Passed in like so: self.json(only_include_changes=True) This method is an easy way to change the comparison logic. `new_value` could be `xyn_types.default.Default`, to indicate that a value's absence is significant (ie: to remove an attribute from destination). Most of the time, a value's absence does not affect the destination when object is sent to API/Service because whatever value is currently there for the attribute is left intact/alone. But sometimes a service will remove the attribute if it does not exist. When this is the case, the absence of the value is significant for comparison purposes; ie: when deciding if a value has changed. :param new_value: New value that will be put into JSON. :param old_value: Old value originals in original JSON [normalized if possible to the same type as new_value. :param field: Field name. :return: If True: Will include the fields value in an update. If False: Won't include the fields value in an update. """ # Convert old value to set if new value is set and old value is list (from original JSON). # If I was really cool :)... I would find out the inner type in case of int/str # and to a conversion to compare Apples to Apples..... # But trying to minimize changes so I don't conflict as much with soon to be # xdynamo feature. if type(new_value) is set and type(old_value) is list: old_value = set(old_value) return new_value != old_value
def update_from_json(self, json: Union[Dict[str, Any], collections.abc.Mapping])
-
REQUIRES associated model object [see self.model].
todo: Needs more documentation
We update the dict per-key, with what we got passed in [via 'json' parameter] overriding anything we got previously. This also makes a copy of the dict, which is want we want [no change to modify the incoming dict parameter].
Expand source code
def update_from_json(self, json: Union[JsonDict, Mapping]): """ REQUIRES associated model object [see self.model]. todo: Needs more documentation We update the dict per-key, with what we got passed in [via 'json' parameter] overriding anything we got previously. This also makes a copy of the dict, which is want we want [no change to modify the incoming dict parameter]. """ structure = self.structure model = self.model api_state = self._api_state if not isinstance(json, Mapping): raise XModelError( f"update_from_json(...) was given a non-mapping parameter ({json})." ) # Merge the old values with the new values. api_state.last_original_update_json = { **(api_state.last_original_update_json or {}), **json } fields = structure.fields values = {} for field_obj in fields: path_list = field_obj.json_path.split(field_obj.json_path_separator) v = json got_value = True for name in path_list: if name not in v: # We don't even have a 'None' value so we assume we just did not get the value # from the api, and therefore we just skip doing anything with it. got_value = False break v = v.get(name) if v is None: break # We map the value we got from JSON into a flat-dict with the BaseModel name as the # key... if got_value: values[field_obj.name] = v if v is not None else Null def set_attr_on_model(field, value, model=model): """ Closure to set attr on self unless value is None. """ if value is None: return setattr(model, field, value) # Merge in the outer json, keeping the values we mapped [via Field.json_path] for conflicts values = {**json, **values} # todo: If the json does not have a value [not even a 'None' value], don't update? # We may have gotten a partial update? For now, always update [even to None] # all defined fields regardless if they are inside the json or not. for field_obj in fields: # We deal with related types later.... if field_obj.related_type: continue f = field_obj.name v = values.get(f, Default) # A None from JSON means a Null for us. # If JSON does not include anything, that's a None for us. if v is None: v = Null elif v is Default: v = None # Run the converter if needed. # If we have a None value, we don't need to convert that, there was no value to # convert. if field_obj.converter and v is not None: v = field_obj.converter( self, Converter.Direction.from_json, field_obj, v ) set_attr_on_model(f, v) for field_obj in fields: # Ok, now we deal with related types... related_type = field_obj.related_type if not related_type: continue f = field_obj.name # todo: at some point, allow customization of this via Field class # Also, s tore the id f_id_name = f"{f}_id" if typing_inspect.get_origin(field_obj.type_hint) is list: # todo: This code is not complete [Kaden never finished it up] # for now, just comment out. raise NotImplementedError( "Type-hints for xmodel models in this format: `attr: List[SomeType]` " "are not currently supported. We want to support it someday. For now you " "must use lower-cased non-generic `list`. At some point the idea is to " "allow one to do `List[ChildModel]` and then we know it's a list of " "other BaseModel objects and automatically handle that in some way." ) # child_type: 'Type[M]' # child_type = typing_inspect.get_args(obj_type) # # __args__ returns a tuple of all arguments passed into List[] so we need to # # pull the class out of the tuple # if child_type: # child_type = child_type[0] # # child_api: BaseApi # child_api = child_type.api # if not child_api and child_api.structure.has_id_field: # # TODO: add a non generic Exception for this # raise XModelError( # f"{model} has an attribute with name ({f}) with type-hint List that " # f"doesn't contain an API BaseModel Type as the only argument" # ) # parent_name = model.__class__.__name__.lower() # state.set_related_field_id(f, parent_name) # continue v = None if f in values: v = values.get(f, Null) if v is not Null: v = related_type(v) # Check to see if we have an api/json field for object relation name with "_id" on # end. if v is None and related_type.api.structure.has_id_field(): # If we don't have a defined field for this value, check JSON for it and store it. # # If we have a defined None value for the id field, meaning the field exists # in the json, and is set directly to None, then we have a Null relationship. # We set that as the value, since there is no need to 'lookup' a null value. f_id_value = json.get(f_id_name) id_field = structure.get_field(f_id_name) if not id_field: id_field = field_obj.related_type.api.structure.get_field('id') # Run the converter if needed. # If we have a None value, we don't need to convert that, there was no value to # convert. if id_field and id_field.converter and f_id_value is not None: f_id_value = id_field.converter( self, Converter.Direction.from_json, id_field, f_id_value ) if f_id_value is None and f_id_name in json: # We have a Null situation. f_id_value = Null if f_id_value is not None: # We have an id! # Set the value to support automatic lookup of value, lazily. # This method also takes care to set child object to Null or delete it # as needed depending on the f_id_value and what the child's id field value is. api_state.set_related_field_id(f, f_id_value) else: # 'v' is either going to be None, Null or an BaseModel object. set_attr_on_model(f, v)
class BaseModel (*args, **initial_values)
-
Used as the abstract base-class for classes/object that communicate with our REST API.
This is one of the main classes, and it's highly recommend you read the SDK Library Overview first, if you have not already. That document has many basic examples of using this class along with other related classes.
Attributes that start with
_
or don't have a type-hint are not considered fields on the object that automatically get mapped to/from the JSON that is passed in. For more details see Type Hints.When you sub-class
BaseModel
, you can create your own Model class, with your own fields/attrs. You can pass class arguments/paramters in when you declare your sub-class. The Model-subclass can provide parameters to the super class during class construction.In the example below, notice the
base_url
part. That's a class argument, that is used by the super-class during the construction of the sub-class (before any instances are created). In this case it takes this and stores it onxmodel.rest.RestStructure.base_model_url
as part of the structure information for theBaseModel
subclass.See Basic Model Example for an example of what class arguments are or look at this example below using a RestModel:
>>> # 'base_url' part is a class argument: >>> from xmodel.rest import RestModel >>> class Account(RestModel["Account"], base_url='/account'): >>> id: str >>> name: str
These class arguments are sent to a special method
BaseStructure.configure_for_model_type()
. See that methods docs for a list of avaliable class-arguments.See
BaseModel.__init_subclass__
for more on the internal details of how this works exactly.Note: In the case of
base_url
example above, it's the base-url-endpoint for the model.If you want to know more about that see
xmodel.rest.RestClient.url_for_endpoint
. It has details on how the final requestUrl
is constructed.This class also allows you to more easily with with JSON data via:
BaseApi.json()
BaseApi.update_from_json()
- Or passing a JSON dict as the first arrument to
BaseModel
.
Other important related classes are listed below.
BaseApi
Accessable viaBaseModel.api
.xmodel.rest.RestClient
: Accessable viaxmodel.base.api.BaseApi.client
.xmodel.rest.settings.RestSettings
: Accessable viaxmodel.base.api.BaseApi.settings
.BaseStructure
: Accessable viaBaseApi.structure
xmodel.base.auth.BaseAuth
: Accessable viaxmodel.base.api.BaseApi.auth
Tip: For all of the above, you can change what class is allocated for each one
by changing the type-hint on a subclass.
Creates a new model object. The first/second params need to be passed as positional arguments. The rest must be sent as key-word arguments. Everything is optional.
Args
id
- Specify the
BaseModel.id
attribute, if you know it. If left as Default, nothing will be set on it. It could be set to something via args[0] (ie: a JSON dict). If you do provide a value, it be set last after everything else has been set. *args
-
I don't want to take names from what you could put into 'initial_values', so I keep it as position-only *args. Once Python 3.8 comes out, we can use a new feature where you can specify some arguments as positional-only and not keyword-able.
FirstArg - If Dict:
If raw dictionary parsed from JSON string. It just calls
self.api.update_from_json(args[0])
for you.FirstArt - If BaseModel:
If a
BaseModel
, will copy fields over that have the same name. You can use this to duplicate a Model object, if you want to copy it. Or can be used to copy fields from one model type into another, on fields that are the same name.Will ignore fields that are present on one but not the other. Only copy fields that are on both models types.
**initial_values
- Let's you specify other attribute values for convenience.
They will be set into the object the same way you would normally doing it:
ie:
model_obj.some_attr = v
is the same asModelClass(some_attr=v)
.
Expand source code
@model_auto_init() # noqa - This will be defined. class BaseModel(ABC): """ Used as the abstract base-class for classes/object that communicate with our REST API. This is one of the main classes, and it's highly recommend you read the [SDK Library Overview](./#orm-library-overview) first, if you have not already. That document has many basic examples of using this class along with other related classes. Attributes that start with `_` or don't have a type-hint are not considered fields on the object that automatically get mapped to/from the JSON that is passed in. For more details see [Type Hints](./#type-hints). When you sub-class `BaseModel`, you can create your own Model class, with your own fields/attrs. You can pass class arguments/paramters in when you declare your sub-class. The Model-subclass can provide parameters to the super class during class construction. In the example below, notice the `base_url` part. That's a class argument, that is used by the super-class during the construction of the sub-class (before any instances are created). In this case it takes this and stores it on `xmodel.rest.RestStructure.base_model_url` as part of the structure information for the `BaseModel` subclass. See [Basic Model Example](./#basic-model-example) for an example of what class arguments are or look at this example below using a RestModel: >>> # 'base_url' part is a class argument: >>> from xmodel.rest import RestModel >>> class Account(RestModel["Account"], base_url='/account'): >>> id: str >>> name: str These class arguments are sent to a special method `xmodel.base.structure.BaseStructure.configure_for_model_type`. See that methods docs for a list of avaliable class-arguments. See `BaseModel.__init_subclass__` for more on the internal details of how this works exactly. .. note:: In the case of `base_url` example above, it's the base-url-endpoint for the model. If you want to know more about that see `xmodel.rest.RestClient.url_for_endpoint`. It has details on how the final request `xurls.url.Url` is constructed. This class also allows you to more easily with with JSON data via: - `xmodel.base.api.BaseApi.json` - `xmodel.base.api.BaseApi.update_from_json` - Or passing a JSON dict as the first arrument to `BaseModel.__init__`. Other important related classes are listed below. - `xmodel.base.api.BaseApi` Accessable via `BaseModel.api`. - `xmodel.rest.RestClient`: Accessable via `xmodel.base.api.BaseApi.client`. - `xmodel.rest.settings.RestSettings`: Accessable via `xmodel.base.api.BaseApi.settings`. - `xmodel.base.structure.BaseStructure`: Accessable via `xmodel.base.api.BaseApi.structure` - `xmodel.base.auth.BaseAuth`: Accessable via `xmodel.base.api.BaseApi.auth` .. tip:: For all of the above, you can change what class is allocated for each one by changing the type-hint on a subclass. """ # ------------------------------------- # --------- Public Properties --------- api: "BaseApi[Self]" = None """ Used to access the api class, which is used to retrieve/send objects to/from api. You can specify this as a type-hint in subclasses to change the class we use for this automatically, like so:: from xmodel import BaseModel, BaseApi from xmodel.base.model import Self from typing import TypeVar M = TypeVar("M") class MyCoolApi(BaseApi[M]): pass class MyCoolModel(BaseModel): # The `Self` here is imported from xmodel # (to maintain backwards compatability with Python < 3.11) # It allows us to communicate our subclass-type to the `api` object, # allowing IDE to type-complete/hint better. api: MyCoolApi[Self] The generic ``T`` type-var in this case refers to whatever model class that your using. In the example just above, ``T`` would be referring to ``MyCoolModel`` if you did this somewhere to get the BaseModel's api: ``MyCoolModel.api``. """ # -------------------------------------------- # --------- Config/Option Properties --------- def __init_subclass__( cls: Type[M], *, lazy_loader: Callable[[Type[M]], None] = None, **kwargs ): """ We take all arguments (except `lazy_loader`) passed into here and send them to the method on our structure: `xmodel.base.structure.BaseStructure.configure_for_model_type`. This allows you to easily configure the BaseStructure via class arguments. For a list of class-arguments, see method parameters for `xmodel.base.structure.BaseStructure.configure_for_model_type`. See [Basic BaseModel Example](./#basic-model-example) for an example of what class arguments are for `BaseModel` classes and how to use them. We lazily configure BaseModel sub-classes. They are configured the first time that `BaseModel.api` is accessed under that subclass. At that point all parent + that specific subclass are configured and an `xmodel.base.api.BaseApi` object is created and set on the `BaseModel.api` attribute. From that point forward, that object is what is used. This only happens the first time that `BaseModel.api` is accessed. If you want to add support for additional BaseModel class arguments, you can do it by modifying the base-implementation `xmodel.base.structure.BaseStructure`. Or if you want it only for a specific sub-set of Models, you can make a custom `xmodel.base.structure.BaseStructure` subclass. You can configure your BaseModel to use this BaseStructure subclass via a type-hint on `xmodel.base.api.BaseApi.structure`. See `xdynamo.dynamo.DynStructure.configure_for_model_type` for a complete example of a custom BaseStructure subclass that adds extra class arguments that are specific to Dynamo. Args: lazy_loader: This is a callable where the first argument is `cls/self`. This is an optional param. If provided, we will call when we need to lazy-load but before we do our normal lazy-loading ourselves here. Most of the time, you'll want to import into the global/module space of where your class lives any imports you need to do lazily, such as circular imports. Right after we call your lazy_loader callable, we will be ourselves calling the method `get_type_hints` to get all of our type-hints. You'll want to be sure all of your forward-referenced type-hints on your model sub-class are resolvable. Forward-ref type hints are the ones that are string based type-hints, they get resolved lazily after your lazy_loader (if provided) is called. You can see in the code in our method below, look at the check for: >>> if "BaseApi" not in globals(): Look at that for a real-world example of what I am talking about. This little piece of code lazily resolves the `BaseApi` type. """ # We are taking all args and sending them to a xmodel.base.structure.BaseStructure # class object. super().__init_subclass__() def lazy_setup_api(cls_or_self): # If requested, before we do our own lazy-loading below, call passed in lazy-loader. if lazy_loader: lazy_loader(cls) for parent in cls.mro(): if parent is not cls and issubclass(parent, BaseModel): # Ensure that parent-class has a chance to lazy-load it's self # before we try to examine our type-hints. getattr(parent, 'api') # We potentially get called a lot (for every sub-class) # so check to see if we already loaded BaseApi type or not. if 'BaseApi' not in globals(): # Lazy import BaseApi into module, helps resolve BaseApi forward-refs; # ie: `api: "BaseApi[T]"` # We need to resolve these due to use of `get_type_hints()` below. # # Sets it in such a way so IDE's such as pycharm don't get confused + pydoc3 # can still find it and use the type forward-reference. # # todo: figure out why dynamic model attribute getter is having an issue with this. # (see that near start of this file at top ^^^) from xmodel import BaseApi globals()['BaseApi'] = BaseApi try: all_type_hints = get_type_hints(cls) except (NameError, AttributeError) as e: from xmodel import XModelError raise XModelError( f"Unable to construct model subclass ({cls}) due to error resolving " f"type-hints on model class. They must be visible at the module-level that " f"the class is defined in. Original error: ({e})." ) from None api_cls: Type["BaseApi"] = all_type_hints['api'] base_cls = None for b in cls.__bases__: if b is BaseModel: break if issubclass(b, BaseModel): base_cls = b break base_api = None if base_cls: base_api = base_cls.api api = api_cls(api=base_api) cls.api = api # Configure structure for our model type with the user supplied options + type-hints. structure = api.structure try: structure.configure_for_model_type( model_type=cls, type_hints=all_type_hints, **kwargs ) except TypeError as e: from xmodel import XModelError # Adding some more information to the exception. raise XModelError( f"Unable to configure model structure for ({cls}) due to error ({e}) " f"while calling ({structure}.configure_for_model_type)." ) return api # The LazyClassAttr will turn into the proper type automatically when it's first accessed. lazy_api = LazyClassAttr(lazy_setup_api, name="api") # Avoids IDE from using this as type-hint for `self.api`, we want it to use the type-hint # defined on attribute. # Otherwise it will try to be too cleaver by trying to use the type in `lazy_api` instead. # The object in `lazy_api` will transform into what has been type-hinted # when it's first accessed by something. setattr(cls, "api", lazy_api) # ------------------------------- # --------- Init Method --------- # todo: Python 3.8 has support for positional-arguments only, do that when we start using it. # See Doc-Comment for what *args is. def __init__(self, *args, **initial_values): """ Creates a new model object. The first/second params need to be passed as positional arguments. The rest must be sent as key-word arguments. Everything is optional. Args: id: Specify the `BaseModel.id` attribute, if you know it. If left as Default, nothing will be set on it. It could be set to something via args[0] (ie: a JSON dict). If you do provide a value, it be set last after everything else has been set. *args: I don't want to take names from what you could put into 'initial_values', so I keep it as position-only *args. Once Python 3.8 comes out, we can use a new feature where you can specify some arguments as positional-only and not keyword-able. ## FirstArg - If Dict: If raw dictionary parsed from JSON string. It just calls `self.api.update_from_json(args[0])` for you. ## FirstArt - If BaseModel: If a `BaseModel`, will copy fields over that have the same name. You can use this to duplicate a Model object, if you want to copy it. Or can be used to copy fields from one model type into another, on fields that are the same name. Will ignore fields that are present on one but not the other. Only copy fields that are on both models types. **initial_values: Let's you specify other attribute values for convenience. They will be set into the object the same way you would normally doing it: ie: `model_obj.some_attr = v` is the same as `ModelClass(some_attr=v)`. """ args_len = len(args) if args_len > 1: raise NotImplementedError( "Passing XContext via second positional argument is no longer supported." ) cls_api_type = type(type(self).api) api = cls_api_type(model=self) setattr(self, "api", api) # Avoids IDE from using this as type-hint for `self.api`. first_arg = args[0] if args_len > 0 else None if isinstance(first_arg, str): # We assume `str` is a json-string, parse json and import. json_objs = json.loads(first_arg) api.update_from_json(json_objs) elif isinstance(first_arg, BaseModel): # todo: Probably make this recursive, in that we copy sub-base-models as well??? api.copy_from_model(first_arg) elif isinstance(first_arg, Mapping): api.update_from_json(first_arg) elif first_arg is not None: raise XModelError( f"When a first argument to BaseModel.__init__ is provided, it needs to be a " f"mapping/dict with the json values in it " f"OR a BaseModel instance to copy from " f"OR a str with a json dict/obj to parse inside of string; " f"I was given a type ({type(first_arg)}) with value ({first_arg}) instead." ) for k, v in initial_values.items(): if not self.api.structure.get_field(k): raise XModelError( f"While constructing {self}, init method got a value for an " f"unknown field ({k})." ) setattr(self, k, v) def __repr__(self): msgs = [] for attr in self.api.list_of_attrs_to_repr(): msgs.append(f'{attr}={getattr(self, attr, None)}') full_message = ", ".join(msgs) return f"{self.__class__.__name__}({full_message})" def __setattr__(self, name, value): # This gets called for every attribute set. # DO NOT use hasattr() in here, because you could make every lazily loaded object load up # [ie: an API request to grab lazily loaded object properties] when the lazy object is set. api = self.api structure = api.structure field = structure.get_field(name) type_hint = None # By default, we go through the special set/converter logic. do_default_attr_set = False if inspect.isclass(self): # If we are a class, just pass it along do_default_attr_set = True elif name == "api": # Don't do anything special with the 'api' var. do_default_attr_set = True elif name.startswith("_"): # don't do anything with private vars do_default_attr_set = True elif name.endswith("_id") and structure.is_field_a_child(name[:-3], and_has_id=True): # We have a virtual field for a related field id, redirect to special setter. state = _private.api.get_api_state(api) state.set_related_field_id(name[:-3], value) return if do_default_attr_set: # We don't do anything more if it's a special attribute/field/etc. super().__setattr__(name, value) return field = structure.get_field(name) if not field: # We don't do anything more without a field object # (ie: just a normal python attribute of some sort, not tied with API). super().__setattr__(name, value) return try: # We have a value going to an attributed that has a type-hint, checking the type... # We will also support auto-converting to correct type if needed and possible, # otherwise an error will be thrown if we can't verify type or auto-convert it. type_hint = field.type_hint value_type = type(value) # todo: idea: We could cache some of these details [perhaps even using closures] # or use dict/set's someday for a speed improvement, if we ever need to. hint_union_sub_types = () if typing_inspect.is_union_type(type_hint): # Gets all args in a union type, to see if one of them will match type_hint. hint_union_sub_types = typing_inspect.get_args(type_hint) # Get first type hint in untion, Field object (where we just got type-hint) # already unwraps the type hint, removing any Null/None types. It's a Union # only if there are other non-Null/None types in a union. For right now # lets only worry about the first one. type_hint = hint_union_sub_types[0] state = _private.api.get_api_state(api) if ( # If we have a blank string, but field is not of type str, # and field is also nullable; we then we convert the value into a Null. # (ie: user is setting a blank-string on a non-string field) field.nullable and type_hint not in [str, None] and value_type is str and not value ): value = Null elif value is None: # By default, this is None [unless user specified something]. value = _get_default_value_from_field(self, field) elif ( value_type is type_hint or value_type in hint_union_sub_types or Optional[value_type] is type_hint or type_hint is NullType and field.nullable ): # Type is the same as type hint, no need to do anything else. # We check to reset any related field id info, just in case it exists, # since this field is either being set to Null or an actual object. state.reset_related_field_id_if_exists(name) pass elif value is Null: # If type_hint supported the Null type, then it would have been dealt with in # the previous if statement. if not field.nullable: raise AttributeError( f"Setting a Null value for field ({name}) when typehint ({type_hint}) " f"does not support NullType, for object ({self})." ) elif field.converter: # todo: Someday map str/int/bool (basic conversions) to standard converter methods; # kind of like I we do it for date/time... have some default converter methods. # # This handles datetime, date, etc... value = field.converter(api, Converter.Direction.to_model, field, value) elif type_hint in (dict, JsonDict) and value_type in (dict, JsonDict): # this is fine for now, keep it as-is! # # For now, we just assume things in the dict are ok. # in the future, we will support `Dict[str, int]` or some such and we will # check/convert/ensure the types as needed. pass elif type_hint in (dict, JsonDict) and value_type in (int, bool, str): # just passively convert bool/int/str into empty dict if type-hint is a dict. log.warning( f"Converted a int/bool/str into an empty dict. Attr name ({name})," f"value ({value}) type-hint ({type_hint}) object ({self}). If you don't want" f"to do this, then don't put a type-hint on the attribute." ) value = {} elif typing_inspect.get_origin(type_hint) in (list, set): # See if we have a converter for this type in our default-converters.... inside_type_hint = typing_inspect.get_args(type_hint)[0] basic_type_converter = self.api.default_converters.get(inside_type_hint) if basic_type_converter: converted_values = [ basic_type_converter( api, Converter.Direction.to_model, field, x ) for x in loop(value) ] container_type = typing_inspect.get_origin(type_hint) value = container_type(converted_values) # Else/Otherwise we just leave things as-is for now, no error and no conversion pass # Python 3.7 does not have GenericMeta anymore, not sure if we need it, we just need # to try using this for a bit and see what happens. # # If needed in Python 3.7, we can see if we can remove this loop-hole with the new # typing_inspect.* methods. # # elif type(type_hint) is GenericMeta: # # This is a complex type (probably a Parameterized generic), not going to try and # # check it out, don't want to throw and error as well, just pass it though. # # # # pass else: raise AttributeError( f"Setting name ({name}) with value ({value}) with type ({value_type}) on " f"API object ({self}) but type-hint is ({type_hint}), and I don't know how" f" to auto-convert type ({value_type}) into ({type_hint})." ) except ValueError: # We want to raise a more informative error than the base ValueError when there # is a problem parsing a value raise AttributeError( f"Parsing value ({value}) with type-hint ({type_hint}) resulted in an error " f"for attribute ({name}) on object ({self})" ) if isinstance(value, str): # This value has caused me a lot of problems, it's time to ALWAYS treat them # as blank strings, exactly what they should have been set to in the first place. if value.startswith('#########'): value = '' if field.post_filter: value = field.post_filter(api=api, name=name, value=value) if field.fset: field.fset(self, value) elif field.fget: raise XModelError( f"We have a field ({field}) that does not have a Field.fset (setter function)," f"but has a Field.fget ({field.fget}) and someone is attempting to set a " f"value on the Model object ({self})... this is unsupported. " f"If you want to allow setting the value, you must provider a setter when a " f"getter is present/provided." ) else: super().__setattr__(name, value) def __getattr__(self, name: str): # Reminder: This method only gets called if attribute is not currently defined in self. structure = self.api.structure state = _private.api.get_api_state(self.api) field = structure.get_field(name) if name.startswith("_"): return object.__getattribute__(self, name) if field and field.fget: # Use getter to get value, if we get a non-None value return it. # If we get a None back, then do the default thing we normally do # (ie: look for default value, related object, etc). value = field.fget(self) if value is not None: return value if name.endswith("_id") and structure.is_field_a_child(name[:-3], and_has_id=True): # We have a field that ends with _id, that when taken off is a child field that # uses and id. This means we should treat this field as virtually related field id. value = state.get_related_field_id(name[:-3]) if value is not None: if field and field.fset: field.fset(self, value) return value if not field: raise AttributeError( f"Getting ({name}) on ({self.__class__.__name__}), which does not exist on object " f"or in API. For API objects, you need to use already defined fields/attributes." ) if ( field.related_type is not None and field.related_type.api.structure.has_id_field() ): name_id_value = state.get_related_field_id(name) # RemoteModel is an abstract interface, # Let's us know how to lazily lookup remote objects by their id value. name_type: "RemoteModel" = field.related_type obj = None if name_id_value is Null: # Don't attempt to lookup object, we have it's primary id: obj = Null elif name_id_value is not None: # Attempt to lookup remote object, we have it's primary id: obj = name_type.api.get_via_id(name_id_value) # todo: consider raising an exception if we have an `id` but no related object? # I'll think about it. # if we have an object, set it and return it if obj is not None: if field.fset: field.fset(self, obj) else: super().__setattr__(name, obj) return obj # Otherwise, we continue, next thing to do is look for any default value. # We next look for a default value, if any set/return that. default = _get_default_value_from_field(self, field) # We set to default value and return it if we have a non-None value. if default is not None: # We have a value of some sort, call setter: if field.fset: field.fset(self, default) else: super().__setattr__(name, default) return default # We found no default value, return None. return None def __eq__(self, other): """ For BaseModel, by default our identity is based on object instance ID, not any values in our attributes. Makes things simpler when trying to find object/self in a Set; which is useful when traversing relationships. """ return self is other def __hash__(self): """ For BaseModel, by default our identity is based on object instance ID, not any values in our attributes. Makes things simpler when trying to find object/self in a Set; which is useful when traversing relationships. """ return id(self)
Ancestors
- abc.ABC
Subclasses
Class variables
var api : BaseApi[BaseModel]
-
Used to access the api class, which is used to retrieve/send objects to/from api.
You can specify this as a type-hint in subclasses to change the class we use for this automatically, like so:: from xmodel import BaseModel, BaseApi from xmodel.base.model import Self from typing import TypeVar
M = TypeVar("M") class MyCoolApi(BaseApi[M]): pass class MyCoolModel(BaseModel): # The <code>Self</code> here is imported from xmodel # (to maintain backwards compatability with Python < 3.11) # It allows us to communicate our subclass-type to the <code><a title="xmodel.base.api" href="api.html">xmodel.base.api</a></code> object, # allowing IDE to type-complete/hint better. api: MyCoolApi[Self]
The generic
T
type-var in this case refers to whatever model class that your using. In the example just above,T
would be referring toMyCoolModel
if you did this somewhere to get the BaseModel's api:MyCoolModel.api
.
class BaseStructure (*, parent: Optional[ForwardRef('BaseStructure')], field_type: Type[~F])
-
BaseStructure class is meant to keep track of things that apply for all
BaseModel
's at the class-level.You can use
BaseStructure.fields
to get all fields for a particularBaseModel
as an example of the sort of information on theBaseStructure
object.BaseStructure is lazily configured for a particular BaseModel the first time something attempts to get
BaseModel.api
off the particular BaseModel subclass.You can get it via first getting api attribute for BaseModel via
BaseModel.api
and then getting the structure attribute on that viaBaseApi.structure
.Example getting the structure object for the Account model/api:
>>> from some_lib.account import Account >>> structure = Account.api.structure
Expand source code
class BaseStructure(Generic[F]): """ BaseStructure class is meant to keep track of things that apply for all `xmodel.base.model.BaseModel`'s at the class-level. You can use `BaseStructure.fields` to get all fields for a particular `xmodel.base.model.BaseModel` as an example of the sort of information on the `BaseStructure` object. BaseStructure is lazily configured for a particular BaseModel the first time something attempts to get `xmodel.base.model.BaseModel.api` off the particular BaseModel subclass. You can get it via first getting api attribute for BaseModel via `xmodel.base.model.BaseModel.api` and then getting the structure attribute on that via `xmodel.base.api.BaseApi.structure`. Example getting the structure object for the Account model/api: >>> from some_lib.account import Account >>> structure = Account.api.structure """ def __init__( self, *, parent: Optional['BaseStructure'], field_type: Type[F] ): super().__init__() # Set specific ones so I have my own 'instance' of them. self._name_to_type_hint_map = {} self._get_fields_cache = None # Copy all my attributes over from parent, for use as 'default' values. if parent: self.__dict__.update(parent.__dict__) # noinspection PyProtectedMember # This parent is my own type/class, so I am fine accessing it's private member. self._name_to_type_hint_map = parent._name_to_type_hint_map.copy() self._get_fields_cache = None self.field_type = field_type self.internal_shared_api_values = {} def configure_for_model_type( self, *, # <-- means we don't support positional arguments model_type: Type['BaseModel'], type_hints: Dict[str, Any], ): """ This EXPECTS to have passed-in the type-hints for my `BaseStructure.; see code inside `xmodel.base.model.BaseModel.__init_subclass__` for more details. There is no need to get the type-hints twice [it can be a bit expensive, trying to limit how may times I grab them].... See `xmodel.base.model.BaseModel` for more details on how Models work... This describes the options you can pass into a `xmodel.base.model.BaseModel` subclass at class-construction time. It allows you to customize how the Model class will work. This method will remember the options passed to it, but won't finish constructing the class until someone asks for it's `xmodel.base.model.BaseModel.api` attribute for the first time. This allows you to dynamically add more Field classes if needed. It also makes things import faster as we won't have to fully setup the class unless something tries to use it. Args: model_type (Type[xmodel.base.model.BaseModel]): The model we are associated with, this is what we are configuring ourselves against. type_hints (Dict[str, Any]): List of typehints via Python's `get_type_hints` method; Be aware that `get_type_hints` will try and resolve all type-hints, including ones that are forward references. Make sure these types are available at the module-level by the time `get_type_hints` runs. """ # Prep model class, remove any class Field objects... # These objects have been "moved" into me via `self.fields`. self._name_to_type_hint_map = type_hints self.model_cls = model_type for field_obj in self.fields: field_name = field_obj.name # The default values are inside `field_obj.default` now. # We delete the class-vars, so that `__getattr__` is called when someone attempts # to grab a value from a BaseModel for an attribute that does not directly exist # on the BaseModel subclass so we can do our normal field_obj.default resolution. # If the class keeps the value, it prevents `__getattr__` from being called for # attributes that don't exist directly on the model instance/object; # Python will instead grab and return the value set on the class for that attribute. # # todo/thoughts/brain-storm: # Consider just using __getattribute__ for BaseModel instead of __getattr_... # It's slightly slower but then I could have more flexablity around this... # Thinking of returning the associated field-object if you do # `BaseModelSubClass.some_attr_field` for various purposes.... # Using `__getattribute__` would allow for this.... # just something I have been thinking about... # For example: you could use that field object as a query-key instead of a string # with the field-name... # might be nicer, and get auto-completion that way... not sure, thinking about it. # if field_name in self.model_cls.__dict__: delattr(self.model_cls, field_name) # -------------------------------------- # --------- Environmental Properties --------- model_cls: "Type[BaseModel]" """ The model's class we are defining the structure for. This is typed as some sort of `xmodel.base.model.BaseModel` . This is NOT generically typed anymore, to get much better generically typed version you should use `xmodel.base.api.BaseApi.model_type` to get the BaseModel outside of the `xmodel.structure` module. Using that will give the IDE the correctly typed BaseModel class! """ # -------------------------------------- # --------- General Properties --------- # # Most of these will be set inside __init_subclass__() via associated BaseModel Class. field_type: Type[F] """ Field type that this structure will use when auto-generating `xmodel.fields.Field`'s. User defined Fields on a model-class will keep whatever type the user used. When `xmodel.base.model.BaseModel` class is constructed, and the `BaseStructure` is created, we will check to ensure all user-defined fields inherit from this field_type. That way you can assume any fields you get off this structure object inherit from field_type. """ internal_shared_api_values: Dict[Any, Any] = None """ A place an `xmodel.base.api.BaseApi` object can use to share values BaseModel-class wide (ie: for all BaseModel's of a specific type). This should NOT be used outside of the BaseApi class. For example, ``xmodel.base.api.BaseApi.client` stores it's object lazily here. Users outside of BaseApi class should simply ask it for the client and not try to go behind it's back and get it here. Code/Users outside of `xmodel.base.api.BaseApi` and it's subclasses can't assume anything about what's in this dictionary. This exists for pure-convenience of the `xmodel.base.api.BaseApi` class. """ _name_to_type_hint_map: Dict[str, Any] """ .. deprecated:: v0.2.33 Use `BaseStructure.fields` instead to get a list of the real fields to use. And `xmodel.fields.Field.type_hint` to get the type-hint [don't get it here, keeping this temporary for backwards compatibility]. A map of attribute-name to type-hint type. .. important:: This WILL NOT take into account field-names where the `Field.name` is different then the name of the field on BaseModel the type-hint was assigned to. """ _get_fields_cache: Dict[str, F] = None @property def have_api_endpoint(self) -> bool: """ Right now, a ready-only property that tells you if this BaseModel has an API endpoint. That's determined right now via seeing if `BaseStructure.has_id_field_set()` is True. """ if not self.has_id_field(): return False else: return True def __copy__(self): obj = type(self)(parent=self, field_type=self.field_type) obj.__dict__.update(self.__dict__) obj._name_to_type_hint_map = self._name_to_type_hint_map.copy() obj._get_fields_cache = None return obj def field_exists(self, name: str) -> bool: """ Return `True` if the field with `name` exists on the model, otherwise `False`. """ return name in self.field_map def has_id_field(self): """ Defaults to False, returns True for RemoteStructure, What this property is really saying is if you can do a foreign-key to the related object/model. It may be better at some point in the long-run to rename this field to more indicate that; perhaps the next time we have a breaking-change we need to do for xmodel. For now, we are leaving the name along and hard-coding this to return False in BaseStructure, and to return True in RemoteStructure. """ return False def get_field(self, name: str) -> Optional[F]: """ Args: name (str): Field name to query on. Returns: xmodel.fields.Field: If field object exists with `name`. None: If not field with `name` exists """ if name is None: return None return self.field_map.get(name) @property def fields(self) -> List[F]: """ Returns: List[xmodel.fields.Field]: list of field objects. """ return list(self.field_map.values()) @property def field_map(self) -> Mapping[str, F]: """ Returns: Dict[str, xmodel.fields.Field]: Map of `xmodel.fields.Field.name` to `xmodel.fields.Field` objects. """ cached_content = self._get_fields_cache if cached_content is not None: # Mapping proxy is a read-only view of the passed in dict. # This will LIVE update the mapping if underlying dict changed. return MappingProxyType(cached_content) generated_fields = self._generate_fields() self._get_fields_cache = generated_fields return MappingProxyType(generated_fields) def excluded_field_map(self) -> Dict[str, F]: """ Returns: Dict[str, xmodel.fields.Field]: Mapping of `xmodel.fields.Field.name` to field objects that are excluded (`xmodel.fields.Field.exclude` == `True`). """ return {f.name: f for f in self.fields if f.exclude} def _generate_fields(self) -> Dict[str, F]: """ Goes though object and grabs/generated Field objects and caches them in self. Gives back the definitive list of Field objects. For now keeping this private, but may open it up in the future if sub-classes need to customize how fields are generated. """ full_field_map = {} default_field_type: Type[Field] = self.field_type type_hint_map = self._name_to_type_hint_map model_cls = self.model_cls # todo: Figure out how to put this into/consolidate into # `xmodel.base.api.BaseApi`; and simplify stuff!!! default_converters = getattr(self.model_cls.api, 'default_converters') # todo: default_con ^^^^ make sure we are using it!!!! # Lazy-import BaseModel, we need to check to see if we have a sub-class or not... from xmodel import BaseModel # This will be a collection of any Fields that exist on the parent(s), merged together... base_fields: Dict[str, Field] = {} # go though parent and find any Field objects, grab latest version # which is the one closest to child on a per-field basis... # we exclude it's self [the model we are currently working with]. for base in reversed(model_cls.__mro__[1:]): base: Type[BaseModel] if not inspect.isclass(base): continue if not issubclass(base, BaseModel): continue if not base.api: # `base` is likely xmodel.base.model.BaseModel; and that has no API allocated # to it # at the moment [mostly because the __init_subclasses is only executed on sub's]. # todo: BaseModel is an abstract class... do we really need structure/fields on it? continue # todo: ensure we later on use these and make a new field if needed... base_fields.update(base.api.structure.field_map) for name, type_hint in type_hint_map.items(): # Ignore the 'api' attribute, it's special. if name == 'api': continue # Ignore anything the starts with '_'. if name.startswith("_"): continue # todo: # 1. Get Parent Field's, merge values. # 2. Map all type's and if not map then raise error. # noinspection PyArgumentList field_obj: Field field_value: Field = getattr(model_cls, name, Default) if isinstance(field_value, Field): field_obj = field_value field_value = Default elif field_value is not Default: if not inspect.isclass(field_value) and isinstance(field_value, property): field_obj = default_field_type(fget=field_value.fget, fset=field_value.fset) else: # noinspection PyArgumentList field_obj = default_field_type(default=field_value) else: # noinspection PyArgumentList field_obj = default_field_type() # Name can be overridden, we want to use it to lookup parent field name.... if field_obj.name: name = field_obj.name field_obj.resolve_defaults( name=name, type_hint=type_hint_map.get(name, None), default_converter_map=default_converters, parent_field=base_fields.get(name) ) # Ensure all fields that still have `Default` as their value are resolved to None. field_obj.resolve_remaining_defaults_to_none() # field-object will unwrap the type-hint for us. type_hint = field_obj.type_hint # Name can be overridden, we want to use whatever it says we should be using. name = field_obj.name full_field_map[field_obj.name] = field_obj # If we have a converter, we can assume that will take care of things correctly # for whatever type we have. If we don't have a converter, we only support specific # types; We check here for type-compatibility. from xmodel import BaseModel if ( not field_obj.converter and type_hint not in supported_basic_types and (not inspect.isclass(type_hint) or not issubclass(type_hint, BaseModel)) and typing_inspect.get_origin(type_hint) not in (list, set) ): raise XModelError( f"Unsupported type ({type_hint}) with field-name ({name}) " f"for model-class ({model_cls}) in field-obj ({field_obj})." ) if ( field_obj.json_path and field_obj.json_path != field_obj.name and field_obj.related_type ): XModelError( "Right now obj-relationships can't use the 'json_path' option " "while at the same time being obj-relationships. Must use basic field " "with api_path. " # Copy/Paste from `BaseApi.json`: f"Can't have xmodel.Field on BaseModel with related-type and a json_path " f"that differ at the moment, for field ({field_obj}). " f"It is something I want to support someday; the support is mostly in place " f"already, but it needs some more careful thought, attention and testing " f"before we should allow it. " "Workaround: Make an `{field.name}_id` field next to related field on the " "model. Then, set `json_path` for that `{field.name}_id` field, set it to " "what you want it to be. Finally, set the `{related_field.name}` to " "read_only=True. This allows you to rename the `_id` field used to/from api " "in the JSON input/output, but the Model can have an alternate name for the " "related field. You can see a real-example of this at " "`bigcommerce.api.orders._BcCommonOrderMetafield.order" ) # todo: Provide a 'remove' option in the Field config class. if 'id' not in full_field_map: # Go though and populate the `Field.field_for_foreign_key_related_field` as needed... for k, f in full_field_map.items(): # If there is a relate field name, and we have a field defined for it... # Set it's field_for_foreign_key_related_field so the correct field... # Otherwise generate a field object for this key-field. # # FYI: The `resolve_defaults` call above will always set # field_for_foreign_key_related_field to None. # We then set it to something here if needed. if f.related_field_name_for_id: related_field = full_field_map.get(f.related_field_name_for_id) if related_field: related_field.field_for_foreign_key_related_field = f return full_field_map def id_cache_key(self, _id): """ Returns a proper key to use for `xmodel.base.client.BaseClient.cache_get` and other caching methods for id-based lookup of an object. """ if type(_id) is dict: # todo: Put module name in this key. key = f"{self.model_cls.__name__}" try: sorted_keys = sorted(_id.keys()) except TypeError: sorted_keys = _id.keys() for key_name in sorted_keys: key += f"-{key_name}-{_id[key_name]}" return key else: return f"{self.model_cls.__name__}-id-{_id}" # todo: Get rid of this [only used by Dynamo right now]. Need to use Field instead... def get_unwraped_typehint(self, field_name: str): """ This is now done for you on `xmodel.fields.Field.type_hint`, so you can just grab it directly your self now. Gets typehint for field_name and calls `xmodel.types.unwrap_optional_type` on it to try and get the plain type-hint as best as we can. """ field = self.get_field(field_name) if field is None: return None return field.type_hint def is_field_a_child(self, child_field_name, *, and_has_id=False): """ True if the field is a child, otherwise False. Will still return `False` if `and_has_id` argument is `True` and the related type is configured to not use id via class argument `has_id_field=False` (see `BaseStructure.configure_for_model_type` for more details on class arguments). Won't raise an exception if field does not exist. Args: child_field_name (str): Name of field to check. and_has_id (bool): If True, then return False if related type is not configured to use id. Returns: bool: `True` if this field is a child field, otherwise `False`. """ field = self.get_field(child_field_name) if not field: return False related_type = field.related_type if not related_type: return False related_structure = related_type.api.structure if and_has_id and not related_structure.has_id_field(): return False return True @property def endpoint_description(self): """ Gives some sort of basic descriptive string that contains the path/table-name/etc that basically indicates the api endpoint being used. This is meant for logging and other human readability/debugging purposes. Feel free to change the string to whatever is most useful to know. I expect this to be overridden by the concrete implementation, see examples here: - `xmodel.rest.RestStructure.endpoint_description` - `xmodel.dynamo.DynStructure.endpoint_description` """ return "?"
Ancestors
- typing.Generic
Subclasses
Class variables
var field_type : Type[~F]
-
Field type that this structure will use when auto-generating
xmodel.fields.Field
's. User defined Fields on a model-class will keep whatever type the user used. WhenBaseModel
class is constructed, and theBaseStructure
is created, we will check to ensure all user-defined fields inherit from this field_type.That way you can assume any fields you get off this structure object inherit from field_type.
-
A place an
BaseApi
object can use to share values BaseModel-class wide (ie: for all BaseModel's of a specific type).This should NOT be used outside of the BaseApi class. For example,
`xmodel.base.api.BaseApi.client
stores it's object lazily here. Users outside of BaseApi class should simply ask it for the client and not try to go behind it's back and get it here.Code/Users outside of
BaseApi
and it's subclasses can't assume anything about what's in this dictionary. This exists for pure-convenience of theBaseApi
class. var model_cls : Type[BaseModel]
-
The model's class we are defining the structure for. This is typed as some sort of
BaseModel
. This is NOT generically typed anymore, to get much better generically typed version you should useBaseApi.model_type
to get the BaseModel outside of thexmodel.structure
module. Using that will give the IDE the correctly typed BaseModel class!
Instance variables
var endpoint_description
-
Gives some sort of basic descriptive string that contains the path/table-name/etc that basically indicates the api endpoint being used.
This is meant for logging and other human readability/debugging purposes. Feel free to change the string to whatever is most useful to know.
I expect this to be overridden by the concrete implementation, see examples here:
xmodel.rest.RestStructure.endpoint_description
xmodel.dynamo.DynStructure.endpoint_description
Expand source code
@property def endpoint_description(self): """ Gives some sort of basic descriptive string that contains the path/table-name/etc that basically indicates the api endpoint being used. This is meant for logging and other human readability/debugging purposes. Feel free to change the string to whatever is most useful to know. I expect this to be overridden by the concrete implementation, see examples here: - `xmodel.rest.RestStructure.endpoint_description` - `xmodel.dynamo.DynStructure.endpoint_description` """ return "?"
var field_map : Mapping[str, ~F]
-
Returns
Dict[str, xmodel.fields.Field]
- Map of
xmodel.fields.Field.name
toxmodel.fields.Field
objects.
Expand source code
@property def field_map(self) -> Mapping[str, F]: """ Returns: Dict[str, xmodel.fields.Field]: Map of `xmodel.fields.Field.name` to `xmodel.fields.Field` objects. """ cached_content = self._get_fields_cache if cached_content is not None: # Mapping proxy is a read-only view of the passed in dict. # This will LIVE update the mapping if underlying dict changed. return MappingProxyType(cached_content) generated_fields = self._generate_fields() self._get_fields_cache = generated_fields return MappingProxyType(generated_fields)
var fields : List[~F]
-
Returns: List[xmodel.fields.Field]: list of field objects.
Expand source code
@property def fields(self) -> List[F]: """ Returns: List[xmodel.fields.Field]: list of field objects. """ return list(self.field_map.values())
var have_api_endpoint : bool
-
Right now, a ready-only property that tells you if this BaseModel has an API endpoint. That's determined right now via seeing if
BaseStructure.has_id_field_set()
is True.Expand source code
@property def have_api_endpoint(self) -> bool: """ Right now, a ready-only property that tells you if this BaseModel has an API endpoint. That's determined right now via seeing if `BaseStructure.has_id_field_set()` is True. """ if not self.has_id_field(): return False else: return True
Methods
def configure_for_model_type(self, *, model_type: Type[ForwardRef('BaseModel')], type_hints: Dict[str, Any])
-
This EXPECTS to have passed-in the type-hints for my `BaseStructure.; see code inside
BaseModel.__init_subclass__()
for more details. There is no need to get the type-hints twice [it can be a bit expensive, trying to limit how may times I grab them]....See
BaseModel
for more details on how Models work… This describes the options you can pass into aBaseModel
subclass at class-construction time. It allows you to customize how the Model class will work.This method will remember the options passed to it, but won't finish constructing the class until someone asks for it's
BaseModel.api
attribute for the first time. This allows you to dynamically add more Field classes if needed. It also makes things import faster as we won't have to fully setup the class unless something tries to use it.Args
model_type
:Type[BaseModel]
- The model we are associated with, this is what we are configuring ourselves against.
type_hints
:Dict[str, Any]
- List of typehints via Python's
get_type_hints
method; Be aware thatget_type_hints
will try and resolve all type-hints, including ones that are forward references. Make sure these types are available at the module-level by the timeget_type_hints
runs.
Expand source code
def configure_for_model_type( self, *, # <-- means we don't support positional arguments model_type: Type['BaseModel'], type_hints: Dict[str, Any], ): """ This EXPECTS to have passed-in the type-hints for my `BaseStructure.; see code inside `xmodel.base.model.BaseModel.__init_subclass__` for more details. There is no need to get the type-hints twice [it can be a bit expensive, trying to limit how may times I grab them].... See `xmodel.base.model.BaseModel` for more details on how Models work... This describes the options you can pass into a `xmodel.base.model.BaseModel` subclass at class-construction time. It allows you to customize how the Model class will work. This method will remember the options passed to it, but won't finish constructing the class until someone asks for it's `xmodel.base.model.BaseModel.api` attribute for the first time. This allows you to dynamically add more Field classes if needed. It also makes things import faster as we won't have to fully setup the class unless something tries to use it. Args: model_type (Type[xmodel.base.model.BaseModel]): The model we are associated with, this is what we are configuring ourselves against. type_hints (Dict[str, Any]): List of typehints via Python's `get_type_hints` method; Be aware that `get_type_hints` will try and resolve all type-hints, including ones that are forward references. Make sure these types are available at the module-level by the time `get_type_hints` runs. """ # Prep model class, remove any class Field objects... # These objects have been "moved" into me via `self.fields`. self._name_to_type_hint_map = type_hints self.model_cls = model_type for field_obj in self.fields: field_name = field_obj.name # The default values are inside `field_obj.default` now. # We delete the class-vars, so that `__getattr__` is called when someone attempts # to grab a value from a BaseModel for an attribute that does not directly exist # on the BaseModel subclass so we can do our normal field_obj.default resolution. # If the class keeps the value, it prevents `__getattr__` from being called for # attributes that don't exist directly on the model instance/object; # Python will instead grab and return the value set on the class for that attribute. # # todo/thoughts/brain-storm: # Consider just using __getattribute__ for BaseModel instead of __getattr_... # It's slightly slower but then I could have more flexablity around this... # Thinking of returning the associated field-object if you do # `BaseModelSubClass.some_attr_field` for various purposes.... # Using `__getattribute__` would allow for this.... # just something I have been thinking about... # For example: you could use that field object as a query-key instead of a string # with the field-name... # might be nicer, and get auto-completion that way... not sure, thinking about it. # if field_name in self.model_cls.__dict__: delattr(self.model_cls, field_name)
def excluded_field_map(self) ‑> Dict[str, ~F]
-
Returns
Dict[str, xmodel.fields.Field]
- Mapping of
xmodel.fields.Field.name
to field objects that are excluded (xmodel.fields.Field.exclude
==True
).
Expand source code
def excluded_field_map(self) -> Dict[str, F]: """ Returns: Dict[str, xmodel.fields.Field]: Mapping of `xmodel.fields.Field.name` to field objects that are excluded (`xmodel.fields.Field.exclude` == `True`). """ return {f.name: f for f in self.fields if f.exclude}
def field_exists(self, name: str) ‑> bool
-
Return
True
if the field withname
exists on the model, otherwiseFalse
.Expand source code
def field_exists(self, name: str) -> bool: """ Return `True` if the field with `name` exists on the model, otherwise `False`. """ return name in self.field_map
def get_field(self, name: str) ‑> Optional[~F]
-
Args
name
:str
- Field name to query on.
Returns
xmodel.fields.Field
- If field object exists with
name
. None
- If not field with
name
exists
Expand source code
def get_field(self, name: str) -> Optional[F]: """ Args: name (str): Field name to query on. Returns: xmodel.fields.Field: If field object exists with `name`. None: If not field with `name` exists """ if name is None: return None return self.field_map.get(name)
def get_unwraped_typehint(self, field_name: str)
-
This is now done for you on
xmodel.fields.Field.type_hint
, so you can just grab it directly your self now.Gets typehint for field_name and calls
xmodel.types.unwrap_optional_type
on it to try and get the plain type-hint as best as we can.Expand source code
def get_unwraped_typehint(self, field_name: str): """ This is now done for you on `xmodel.fields.Field.type_hint`, so you can just grab it directly your self now. Gets typehint for field_name and calls `xmodel.types.unwrap_optional_type` on it to try and get the plain type-hint as best as we can. """ field = self.get_field(field_name) if field is None: return None return field.type_hint
def has_id_field(self)
-
Defaults to False, returns True for RemoteStructure, What this property is really saying is if you can do a foreign-key to the related object/model.
It may be better at some point in the long-run to rename this field to more indicate that; perhaps the next time we have a breaking-change we need to do for xmodel.
For now, we are leaving the name along and hard-coding this to return False in BaseStructure, and to return True in RemoteStructure.
Expand source code
def has_id_field(self): """ Defaults to False, returns True for RemoteStructure, What this property is really saying is if you can do a foreign-key to the related object/model. It may be better at some point in the long-run to rename this field to more indicate that; perhaps the next time we have a breaking-change we need to do for xmodel. For now, we are leaving the name along and hard-coding this to return False in BaseStructure, and to return True in RemoteStructure. """ return False
def id_cache_key(self, _id)
-
Returns a proper key to use for
xmodel.base.client.BaseClient.cache_get
and other caching methods for id-based lookup of an object.Expand source code
def id_cache_key(self, _id): """ Returns a proper key to use for `xmodel.base.client.BaseClient.cache_get` and other caching methods for id-based lookup of an object. """ if type(_id) is dict: # todo: Put module name in this key. key = f"{self.model_cls.__name__}" try: sorted_keys = sorted(_id.keys()) except TypeError: sorted_keys = _id.keys() for key_name in sorted_keys: key += f"-{key_name}-{_id[key_name]}" return key else: return f"{self.model_cls.__name__}-id-{_id}"
def is_field_a_child(self, child_field_name, *, and_has_id=False)
-
True if the field is a child, otherwise False. Will still return
False
ifand_has_id
argument isTrue
and the related type is configured to not use id via class argumenthas_id_field=False
(seeBaseStructure.configure_for_model_type()
for more details on class arguments).Won't raise an exception if field does not exist.
Args
child_field_name
:str
- Name of field to check.
and_has_id
:bool
- If True, then return False if related type is not configured to use id.
Returns
bool
True
if this field is a child field, otherwiseFalse
.
Expand source code
def is_field_a_child(self, child_field_name, *, and_has_id=False): """ True if the field is a child, otherwise False. Will still return `False` if `and_has_id` argument is `True` and the related type is configured to not use id via class argument `has_id_field=False` (see `BaseStructure.configure_for_model_type` for more details on class arguments). Won't raise an exception if field does not exist. Args: child_field_name (str): Name of field to check. and_has_id (bool): If True, then return False if related type is not configured to use id. Returns: bool: `True` if this field is a child field, otherwise `False`. """ field = self.get_field(child_field_name) if not field: return False related_type = field.related_type if not related_type: return False related_structure = related_type.api.structure if and_has_id and not related_structure.has_id_field(): return False return True