class DynApi(RemoteApi[M]):
    """
    Put things here that are only relevant for all DynModel's.

    The main change vs the base BaseApi class is filtering out of the JSON invalid blank values.

    Right now this model class is only used to transform json from/to Dynamo via
    the `json` and `update_from_json()` methods. See `xmodel.base.api.BaseApi` for more
    details.
    """
    client: DynClient[M]
    structure: DynStructure[DynField]
    # This type-hint is only for IDE, `RemoteApi` does not use it
    # (self.model_type value is passed in when RemoteApi is allocated, in __init__ method).
    model: M

    def get(
            self,
            query: Query = None,
            *,
            top: int = None,
            fields: FieldNames | DefaultType | None = Default,
            allow_scan: bool = False,
            consistent_read: bool | DefaultType = Default,
            reverse: bool = False
    ) -> Iterable[M]:
        """
        Convenience method for the `self.client.get` method.

        Generally, will get/query/scan/batch-get; generally,
        the client will figure out the best way to get the items based on the provided query.
        If the query is None or blank, it will grab everything in the table.

        Args:
            query: A dict with key the field, with optional double-underscore and operation
                (such as `{'some_field__beginswith': 'search-value'}`).
                The value is what to search for.
                If you give this a list, it implies a `__in` operator, and will do an OR
                on the values in the list.
            top:
            fields:
            allow_scan: Defaults to False, which means this will raise an exception if a scan
                is required to execute your get.  Set to True to allow a scan if needed
                (it will still do a query, batch-get, etc; if it can, it only does a scan
                if there is no other choice).

                If the query is blank or None will still do a scan regardless of what you pass
                (to return all items in the table).
            consistent_read: Defaults to Model.api.structure.dyn_consistent_read_by_default,
                which can be set via class arguments when DynModel subclass is defined.

                You can use this to override the model default. True means we use consistent
                reads, otherwise false.
            reverse: Defaults to False, which means sort order is ascending.
                Set to True to reverse the order, which will set the "ScanIndexForward"
                parameter to False in the query.

        Returns:

        """
        return self.client.get(
            query, top=top, fields=fields, allow_scan=allow_scan, consistent_read=consistent_read, reverse=reverse
        )

    def get_key(self, hash_key: Any, range_key: Optional[Any] = None) -> DynKey:
        """
        Easy way to generate a basic `DynKey` with hash_key, and range_key
        (if model has range key).

        If you don't provide a range-key and the model needs a range-key,
        will raise an `xmodel.remote.errors.XRemoteError`.
        """
        return DynKey(api=self, hash_key=hash_key, range_key=range_key)

    def get_via_id(
            self,
            id: Union[
                    int | str | UUID,
                    List[int | str | UUID],
                    Dict[str, str | int | UUID],
                    List[Dict[str, str | int | UUID]],
            ],
            fields: FieldNames = Default,
            id_field: str = None,
            aux_query: Query = None,
            consistent_read: bool | DefaultType = Default,
    ) -> Union[Iterable[M], M, None]:
        """
        Overridden in DynApi to convert any provided `DynKey` into string-based `id` and
        passing them to super and returning the result.

        See `xmodel.remote.api.RemoteApi.get_via_id` for more details on how this method works.

        Args:
            id: In addition to `str` and `int` values, you can also used `DynKey`(s) if you wish.
            fields: See `xmodel.remote.api.RemoteApi.get_via_id`
            id_field: See `xmodel.remote.api.RemoteApi.get_via_id`
            aux_query: See `xmodel.remote.api.RemoteApi.get_via_id`
            consistent_read: Defaults to Model.api.structure.dyn_consistent_read_by_default,
                which can be set via class arguments when DynModel subclass is defined.

                You can use this to override the model default. True means we use consistent
                reads, otherwise false.

        Returns:
            See `xmodel.remote.api.RemoteApi.get_via_id`
        """
        if id is None:
            return None

        is_list = isinstance(id, list)
        if is_list:
            id: Union[DynKey, int, str]
            new_id = [v.id if type(v) is DynKey else v for v in xloop(id)]
        else:
            new_id = id.id if type(id) is DynKey else id

        if consistent_read is Default:
            return super().get_via_id(new_id, fields=fields, id_field=id_field, aux_query=aux_query)

        # If we have a non-Default consistent-read, then temporarily inject the option.
        dyn_options = DynClientOptions()
        dyn_options.consistent_read = consistent_read
        with dyn_options:
            return super().get_via_id(new_id, fields=fields, id_field=id_field, aux_query=aux_query)

    @property
    def table(self) -> TableResource:
        """ Returns the boto3 table resource to use for our related DynModel.
            Don't cache or hang onto this, it's already properly cached for you via the current
            Context and so will work in every situation [unit-tests, config-changes, etc]...
        """

        if not self.structure.dyn_hash_key_name:
            raise XRemoteError(
                f"While constructing {self.structure.model_cls}, found no hash-key field. "
                f"You must have at least one hash-key field."
            )

        # Look it up each time in case config/service/env/context changes enough
        # for it to be different. DynamoDB will cache the table by name and so it's
        # very fast on subsequent lookups.
        table_name = self.structure.fully_qualified_table_name()

        # noinspection PyTypeChecker
        return DynamoDB.grab().table(name=table_name, table_creator=self._create_table)

    def _create_table(self, dynamo: DynamoDB) -> TableResource:
        return self.client.create_table()

    def delete(self, *, condition: Query = None):
        """
        REQUIRES associated model object [see self.model].

        Convenience method to delete this single object in API.

        If you pass in a condition, it will be evaluated on the dynamodb-service side and the
        deletes will only happen if the condition is met.
        This can help prevent race conditions if used correctly.

        If there is a batch-writer currently in use, we will try to use that to batch the deletes.

        Keep in mind that if you pass in a `condition`, we can't use the batch write.
        We will instead send of a single-item delete request with the condition attached
        (bypassing any current batch-writer that may be in use).

        Args:
            condition: Conditions in query will be sent to dynamodb; object will only be deleted
                if the conditions match.
                See doc comments on `xyn_model_dynamo.client.DynClient.delete_obj` for more
                details.
        """

        # Normally I would call super().delete(...) and have the super have a **kwargs it can
        # simply pass along (so it can do any normal checks it does).
        # Don't want to modify another library right now, so for now copying/pasting the code here.
        model = self.model
        if model.id is None:
            raise XRemoteError(
                f"A delete was requested for an object that had no id for ({model})."
            )

        self.client.delete_obj(model, condition=condition)

    def send(self, *, condition: Query = None):
        """
        """

        # Normally I would call super().delete(...) and have the super have a **kwargs it can
        # simply pass along (so it can do any normal checks it does).
        # Don't want to modify another library right now, so for now copying/pasting the code here.
        model = self.model
        if model.id is None:
            raise XRemoteError(
                f"A send was requested for an object that had no id for ({model})."
            )

        self.client.send_objs([model], condition=condition)

Ancestors

• xmodel.remote.api.RemoteApi
• xmodel.base.api.BaseApi
• typing.Generic

Instance variables

prop client : DynClient[~M]

Returns an appropriate concrete xmodel.remote.client.RemoteClient subclass. We figure out the proper client object to use based on the type-hint for "client" property on the sub-class.

Example

>>> from typing import TypeVar
>>> from xmodel import RestApi, RestClient
>>> M = TypeVar("M")  # <-- This allows IDE to do better code completion.
>>>
>>> class MyClient(RestClient[M]):
>>>     pass
>>>
>>> class MyApi(RestApi[M])
>>>     client: MyClient[M]  # <-- Type hint on 'client' property.

This is enough for xmodel.base.BaseModel subclasses that have this set as their api type-hint:

>>> from xmodel.remote.model import RemoteModel
>>>
>>> class MyModel(RemoteModel):
>>>     api: MyApi

When you get MyModel's api like below, it will return a MyApi instance, MyApi will in turn return a MyClient:

>>> print(MyModel.api)
MyApi(...)
>>> print(MyModel.api.client)
MyClient(...)

For a more concreate use/example, see xmodel_rest.RestModel; it's a RemoteModel subclass that implments a RestClient that can be used with it.

Expand source code

@property
def _client(self):
    """ Returns an appropriate concrete `xmodel.remote.client.RemoteClient` subclass.
        We figure out the proper client object to use based on the type-hint for "client"
        property on the sub-class.

        Example:

            >>> from typing import TypeVar
            >>> from xmodel import RestApi, RestClient
            >>> M = TypeVar("M")  # <-- This allows IDE to do better code completion.
            >>>
            >>> class MyClient(RestClient[M]):
            >>>     pass
            >>>
            >>> class MyApi(RestApi[M])
            >>>     client: MyClient[M]  # <-- Type hint on 'client' property.

        This is enough for `xmodel.base.BaseModel` subclasses that have this set as their
        api type-hint:

            >>> from xmodel.remote.model import RemoteModel
            >>>
            >>> class MyModel(RemoteModel):
            >>>     api: MyApi

        When you get MyModel's api like below, it will return a MyApi instance,
        MyApi will in turn return a MyClient:

            >>> print(MyModel.api)
            MyApi(...)
            >>> print(MyModel.api.client)
            MyClient(...)

        For a more concreate use/example, see `xmodel_rest.RestModel`;
        it's a RemoteModel subclass that implments a RestClient that can be used with it.
    """
    client = self.structure.internal_shared_api_values.get('client')
    if client:
        return client

    client_type = get_type_hints(type(self)).get('client', None)
    if client_type is None:
        raise XModelError(
            f"RemoteClient subclass type is undefined for model class ({self.model_type}), "
            f"a type-hint for 'client' on BaseApi class must be in place for me to know what "
            f"type to get."
        )

    client = client_type(api=self.model_type.api)
    self.structure.internal_shared_api_values['client'] = client
    return client

prop 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

prop structure : DynStructure[DynField]

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.

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

prop table : boto3.dynamodb.table.TableResource

Returns the boto3 table resource to use for our related DynModel. Don't cache or hang onto this, it's already properly cached for you via the current Context and so will work in every situation [unit-tests, config-changes, etc]…

Expand source code

@property
def table(self) -> TableResource:
    """ Returns the boto3 table resource to use for our related DynModel.
        Don't cache or hang onto this, it's already properly cached for you via the current
        Context and so will work in every situation [unit-tests, config-changes, etc]...
    """

    if not self.structure.dyn_hash_key_name:
        raise XRemoteError(
            f"While constructing {self.structure.model_cls}, found no hash-key field. "
            f"You must have at least one hash-key field."
        )

    # Look it up each time in case config/service/env/context changes enough
    # for it to be different. DynamoDB will cache the table by name and so it's
    # very fast on subsequent lookups.
    table_name = self.structure.fully_qualified_table_name()

    # noinspection PyTypeChecker
    return DynamoDB.grab().table(name=table_name, table_creator=self._create_table)

Methods

def delete(self, *, condition: Dict[str, Union[str, int, datetime.date, xurls.url._FormattedQueryValue, ForwardRef(None), uuid.UUID, Iterable[str | int | uuid.UUID | datetime.date | xurls.url._FormattedQueryValue]]] = None)

REQUIRES associated model object [see self.model].

Convenience method to delete this single object in API.

If you pass in a condition, it will be evaluated on the dynamodb-service side and the deletes will only happen if the condition is met. This can help prevent race conditions if used correctly.

If there is a batch-writer currently in use, we will try to use that to batch the deletes.

Keep in mind that if you pass in a condition, we can't use the batch write. We will instead send of a single-item delete request with the condition attached (bypassing any current batch-writer that may be in use).

Args

condition

Conditions in query will be sent to dynamodb; object will only be deleted if the conditions match. See doc comments on xyn_model_dynamo.client.DynClient.delete_obj for more details.

def get(self, query: Dict[str, Union[str, int, datetime.date, xurls.url._FormattedQueryValue, ForwardRef(None), uuid.UUID, Iterable[str | int | uuid.UUID | datetime.date | xurls.url._FormattedQueryValue]]] = None, *, top: int = None, fields: Union[Sequence[str], xsentinels.default.DefaultType, ForwardRef(None)] = Default, allow_scan: bool = False, consistent_read: bool | xsentinels.default.DefaultType = Default, reverse: bool = False) ‑> Iterable[~M]

Convenience method for the self.client.get method.

Generally, will get/query/scan/batch-get; generally, the client will figure out the best way to get the items based on the provided query. If the query is None or blank, it will grab everything in the table.

Args

query

A dict with key the field, with optional double-underscore and operation (such as {'some_field__beginswith': 'search-value'}). The value is what to search for. If you give this a list, it implies a __in operator, and will do an OR on the values in the list.

top:

fields:

allow_scan

Defaults to False, which means this will raise an exception if a scan is required to execute your get. Set to True to allow a scan if needed (it will still do a query, batch-get, etc; if it can, it only does a scan if there is no other choice).

If the query is blank or None will still do a scan regardless of what you pass (to return all items in the table).

consistent_read

Defaults to Model.api.structure.dyn_consistent_read_by_default, which can be set via class arguments when DynModel subclass is defined.

You can use this to override the model default. True means we use consistent reads, otherwise false.

reverse

Defaults to False, which means sort order is ascending. Set to True to reverse the order, which will set the "ScanIndexForward" parameter to False in the query.

Returns:

def get_key(self, hash_key: Any, range_key: Optional[Any] = None) ‑> DynKey

Easy way to generate a basic DynKey with hash_key, and range_key (if model has range key).

If you don't provide a range-key and the model needs a range-key, will raise an xmodel.remote.errors.XRemoteError.

def get_via_id(self, id: Union[int, str, uuid.UUID, List[int | str | uuid.UUID], Dict[str, str | int | uuid.UUID], List[Dict[str, str | int | uuid.UUID]]], fields: Sequence[str] = Default, id_field: str = None, aux_query: Dict[str, Union[str, int, datetime.date, xurls.url._FormattedQueryValue, ForwardRef(None), uuid.UUID, Iterable[str | int | uuid.UUID | datetime.date | xurls.url._FormattedQueryValue]]] = None, consistent_read: bool | xsentinels.default.DefaultType = Default) ‑> Union[Iterable[~M], ~M, ForwardRef(None)]

Overridden in DynApi to convert any provided DynKey into string-based id and passing them to super and returning the result.

See xmodel.remote.api.RemoteApi.get_via_id for more details on how this method works.

Args

id

In addition to str and int values, you can also used DynKey(s) if you wish.

fields

See xmodel.remote.api.RemoteApi.get_via_id

id_field

See xmodel.remote.api.RemoteApi.get_via_id

aux_query

See xmodel.remote.api.RemoteApi.get_via_id

consistent_read

Defaults to Model.api.structure.dyn_consistent_read_by_default, which can be set via class arguments when DynModel subclass is defined.

You can use this to override the model default. True means we use consistent reads, otherwise false.

Returns

See xmodel.remote.api.RemoteApi.get_via_id

def send(self, *, condition: Dict[str, Union[str, int, datetime.date, xurls.url._FormattedQueryValue, ForwardRef(None), uuid.UUID, Iterable[str | int | uuid.UUID | datetime.date | xurls.url._FormattedQueryValue]]] = None)