Repository: xoeye/xoto3 (language: python)

Issue:
feat: TypedTable API for write_versioned, plus `presume` and even lazier batch gets
see the added readme


Discussion / hints:

Fix patch:
diff --git a/CHANGES.md b/CHANGES.md
index 5e23cb1..98cf065 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,4 +1,15 @@
-### 1.11.0
+## 1.12.0
+
+An enhanced API supporting the use of
+`dynamodb.write_versioned.versioned_transact_write_items`, which is a
+great way to write business logic against DynamoDB.
+
+### 1.11.1
+
+Change all Resources and Clients to use a thread-local Session to
+avoid race conditions.
+
+## 1.11.0
 
 - Update `find_index` and `require_index` to allow for an index with no range key.
 
diff --git a/dev-utils/watch_ddb_stream.py b/dev-utils/watch_ddb_stream.py
index b3a1912..4ba19e3 100755
--- a/dev-utils/watch_ddb_stream.py
+++ b/dev-utils/watch_ddb_stream.py
@@ -12,10 +12,6 @@
 from xoto3.dynamodb.streams.records import old_and_new_items_from_stream_event_record
 from xoto3.dynamodb.utils.index import hash_key_name, range_key_name
 
-DDB_RES = boto3.resource("dynamodb")
-
-DDB_STREAMS_CLIENT = boto3.client("dynamodbstreams")
-
 
 def make_accept_stream_item_for_table(item_slicer: Callable[[dict], str]):
     def accept_stream_item(record: dict):
@@ -55,6 +51,10 @@ def main():
     )
     args = parser.parse_args()
 
+    DDB_RES = boto3.resource("dynamodb")
+
+    DDB_STREAMS_CLIENT = boto3.client("dynamodbstreams")
+
     table = DDB_RES.Table(args.table_name)
 
     if args.attribute_names:
diff --git a/xoto3/__about__.py b/xoto3/__about__.py
index d667ad2..10f2830 100644
--- a/xoto3/__about__.py
+++ b/xoto3/__about__.py
@@ -1,4 +1,4 @@
 """xoto3"""
-__version__ = "1.11.0"
+__version__ = "1.12.0"
 __author__ = "Peter Gaultney"
 __author_email__ = "pgaultney@xoi.io"
diff --git a/xoto3/cloudformation/__init__.py b/xoto3/cloudformation/__init__.py
index a9138f8..cf64311 100644
--- a/xoto3/cloudformation/__init__.py
+++ b/xoto3/cloudformation/__init__.py
@@ -1,10 +1,9 @@
 """Gets and caches things from Cloudformation"""
 import typing as ty
 
-from xoto3.lazy_session import tlls
+from xoto3.lazy_session import tll_from_session
 
-
-_CF_RESOURCE = tlls("resource", "cloudformation")
+_CF_RESOURCE = tll_from_session(lambda sess: sess.resource("cloudformation"))
 
 
 _STACKS: ty.Dict[str, ty.Any] = dict()
diff --git a/xoto3/cloudwatch/metrics.py b/xoto3/cloudwatch/metrics.py
index 54bc1f9..b53169f 100644
--- a/xoto3/cloudwatch/metrics.py
+++ b/xoto3/cloudwatch/metrics.py
@@ -11,12 +11,12 @@
 import botocore
 from typing_extensions import TypedDict
 
-from xoto3.lazy_session import tlls
+from xoto3.lazy_session import tll_from_session
 
 logger = getLogger(__name__)
 
 
-CLOUDWATCH_CLIENT = tlls("client", "cloudwatch")
+CLOUDWATCH_CLIENT = tll_from_session(lambda sess: sess.client("cloudwatch"))
 
 
 MetricUnits = frozenset(
diff --git a/xoto3/dynamodb/README.md b/xoto3/dynamodb/README.md
index 5b71992..de663db 100644
--- a/xoto3/dynamodb/README.md
+++ b/xoto3/dynamodb/README.md
@@ -80,25 +80,30 @@ refetching, and eventually giving up to the utility.
 ```python
 import xoto3.dynamodb.write_versioned as wv
 
+user_table = wv.ItemTable('User')
+group_table = wv.ItemTable('Group')
+# ^ the above are convenience APIs and are pure utilities; they perform no
+# IO and contain no mutable state or clients.
+
 user_key = dict(id="bob")
 group_key = dict(pk="team42")
 
+
 def add_user_to_new_or_existing_group(t: wv.VersionedTransaction) -> wv.VersionedTransaction:
-    user = wv.require(t, "User", user_key)
+    user = user_table.require(user_key)(t)
     assert user, "require will raise ItemNotFoundException if the item does not exist"
-    group = wv.get(t, "Group", group_key)
+    group = group_table.get(group_key)(t)
 
     if group_key not in user["groups"]:
         user["groups"].append(group_key)
-        t = wv.put(t, "User", user)
+        t = user_table.put(user)(t)
 
-    if group:
-        if user_key not in group["members"]:
-            group["members"].append(user_key)
-    else:
-        group = dict(group_key, members=[user_key])
+    if not group:
+        group = dict(group_key, members=list())
+    if user_key not in group["members"]:
+        group["members"].append(user_key)
 
-    if group != wv.get(t, "Group", group_key):
+    if group != group_table.get(group_key):
         # if there was a change to the group
         t = wv.put(t, "Group", group)
     return t
@@ -116,3 +121,5 @@ The above code will ensure that the 'state' of the collection of items
 as defined by your function is fully realized without intervening
 transactions, or retried with the new state of those items if the
 transaction is beaten or otherwise interfered with.
+
+For further documentation on this utility, see the full [readme](./write_versioned/README.md)
diff --git a/xoto3/dynamodb/asynch.py b/xoto3/dynamodb/asynch.py
index cf542c1..7c7bbd0 100644
--- a/xoto3/dynamodb/asynch.py
+++ b/xoto3/dynamodb/asynch.py
@@ -1,23 +1,22 @@
 """A module that provides the core functionality for being able to
 write async code that transparently batches requests to Dynamo.
 """
+import asyncio
+import logging
+import traceback
 import typing as ty
+from asyncio import Queue, get_event_loop
 from collections import defaultdict
 from timeit import default_timer
-import traceback
-import logging
 
-import asyncio
-from asyncio import Queue, get_event_loop
+from xoto3.lazy_session import tll_from_session
 
-from xoto3.lazy_session import tlls
 from .batch_get import BatchGetItemTupleKeys
 
-
 logger = logging.getLogger(__name__)
 
 
-DYNAMODB_RESOURCE = tlls("resource", "dynamodb")
+DYNAMODB_RESOURCE = tll_from_session(lambda sess: sess.resource("dynamodb"))
 
 
 # When running in a Lambda, the round-trip latency to Dynamo is
@@ -144,9 +143,7 @@ def table_batch_get_fulfiller(key_tuples: ty.List[tuple]) -> ty.List[dict]:
             table_batch_get_fulfiller, queue, request_map, logging_name=table_name
         )
     )
-    fulfiller = AsyncBatchRequestFulfiller(  # type: ignore
-        queue, request_map, task
-    )
+    fulfiller = AsyncBatchRequestFulfiller(queue, request_map, task)  # type: ignore
     return fulfiller
 
 
diff --git a/xoto3/dynamodb/batch_get.py b/xoto3/dynamodb/batch_get.py
index 2ac5321..6f67034 100644
--- a/xoto3/dynamodb/batch_get.py
+++ b/xoto3/dynamodb/batch_get.py
@@ -1,19 +1,18 @@
 """Utilities for BatchGets from DynamoDB"""
-import typing as ty
-from typing import Iterable, Tuple, Set, List
-import timeit
-from multiprocessing.dummy import Pool as ThreadPool
 import os
+import timeit
+import typing as ty
 from logging import getLogger
+from multiprocessing.dummy import Pool as ThreadPool
+from typing import Iterable, List, Set, Tuple
 
 from xoto3.backoff import backoff
-from xoto3.lazy_session import tlls
+from xoto3.dynamodb.types import TableResource
+from xoto3.lazy_session import tll_from_session
 from xoto3.utils.iter import grouper_it, peek
 from xoto3.utils.lazy import Lazy
-from xoto3.dynamodb.types import TableResource
-
-from .types import Item, KeyTuple, ItemKey, KeyAttributeType
 
+from .types import Item, ItemKey, KeyAttributeType, KeyTuple
 
 logger = getLogger(__name__)
 
@@ -23,7 +22,7 @@
     lambda: ThreadPool(_THREADPOOL_SIZE) if _THREADPOOL_SIZE else None
 )
 
-_DYNAMODB_RESOURCE = tlls("resource", "dynamodb")
+_DYNAMODB_RESOURCE = tll_from_session(lambda sess: sess.resource("dynamodb"))
 
 
 class KeyItemPair(ty.NamedTuple):
diff --git a/xoto3/dynamodb/get.py b/xoto3/dynamodb/get.py
index 3308c2d..b7460aa 100644
--- a/xoto3/dynamodb/get.py
+++ b/xoto3/dynamodb/get.py
@@ -1,4 +1,6 @@
+from functools import wraps
 from logging import getLogger
+from typing import Callable, TypeVar, cast
 
 from .constants import DEFAULT_ITEM_NAME
 from .exceptions import ItemNotFoundException, raise_if_empty_getitem_response
@@ -43,3 +45,22 @@ def strongly_consistent_get_item_if_exists(
         return strongly_consistent_get_item(table, key, nicename=nicename)
     except ItemNotFoundException:
         return dict()
+
+
+F = TypeVar("F", bound=Callable)
+
+
+def retry_notfound_consistent_read(get_item: F) -> F:
+    @wraps(get_item)
+    def get_with_consistent_read_retry(*args, **kwargs):
+        try:
+            return get_item(*args, **kwargs)
+        except ItemNotFoundException:
+            con_read = kwargs.get("ConsistentRead")
+            if con_read:
+                # we already did a consistent read
+                raise
+            logger.info("Retrying with a consistent read")
+            return get_item(*args, **dict(kwargs, ConsistentRead=True))
+
+    return cast(F, get_with_consistent_read_retry)
diff --git a/xoto3/dynamodb/streams/shards.py b/xoto3/dynamodb/streams/shards.py
index c596609..490ae47 100644
--- a/xoto3/dynamodb/streams/shards.py
+++ b/xoto3/dynamodb/streams/shards.py
@@ -1,12 +1,12 @@
 import typing as ty
-from typing_extensions import TypedDict
 
-import boto3
+from typing_extensions import TypedDict
 
+from xoto3.lazy_session import tll_from_session
 from xoto3.paginate import (
-    yield_pages_from_operation,
     DYNAMODB_STREAMS_DESCRIBE_STREAM,
     DYNAMODB_STREAMS_GET_RECORDS,
+    yield_pages_from_operation,
 )
 
 
@@ -31,7 +31,7 @@ class ShardIterator(TypedDict):
 
 class StreamsClient:
     def __init__(self):
-        self.client = boto3.client("dynamodbstreams")
+        self.client = tll_from_session(lambda sess: sess.client("dynamodbstreams"))()
 
     def yield_shards(self, StreamArn: str) -> ty.Iterable[Shard]:
         """This is an expensive operation relatively speaking
diff --git a/xoto3/dynamodb/write_versioned/README.md b/xoto3/dynamodb/write_versioned/README.md
new file mode 100644
index 0000000..df00e6d
--- /dev/null
+++ b/xoto3/dynamodb/write_versioned/README.md
@@ -0,0 +1,132 @@
+# versioned_transact_write_items
+
+This is a general-purpose system for writing retrying DynamoDB
+transactional operations for 1 to
+[25](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Limits.html#limits-dynamodb-transactions)
+items, using pure functions.
+
+The magic sauce here is the versioning - we use ConditionExpressions
+based on a specified attribute, by default `item_version`, to detect
+whether any intervening writes have been made, such that an `update`
+can safely be expressed as a simple `put`. This makes the
+[ABA problem](https://en.wikipedia.org/wiki/ABA_problem) impossible,
+so you can write your business logic free of worry about the
+complexity of transactions in the real world.
+
+The API follows a consistent pattern of using provided functions to
+modify an opaque, immutable VersionedTransaction object. All writes to
+the database are performed by calling a function with the current
+state of the versioned transaction, resulting in a new
+VersionedTransaction object that may be further evolved using other
+provided functions. Crucially, the API exposes no 'state' about the
+transaction except for via simple read and modify operations,
+resulting in a fully composable transaction system - any transaction
+builder can be composed with any other in whatever logical order is
+necessary.
+
+## Low Level API
+
+At base, you will be calling `versioned_transact_write_items`, and
+passing a function which makes use of the VersionedTransaction
+API. The low level implementation of the VersionedTransaction API is
+defined in three files - `read.py`, `modify.py`, and `specify.py`,
+though it is meant to be imported via the top level module.
+
+## High Level Table-oriented API
+
+From use we've discovered that it's nicer to say certain things once
+and not have to repeat ourselves. A higher level table-oriented API is
+now offered via the `ItemTable` and `TypedTable` classes.
+
+This API provides partial-application-by-default, because that makes
+certain use cases much cleaner to express. This means that you'll be
+writing things like `table.get(key)(transaction)` instead of
+`table.get(key, transaction)`, i.e. using more parentheses rather than
+commas.
+
+### put unless exists
+
+As a simple but non-trivial example, you can perform a put-unless-exists like so:
+
+```python
+import xoto3.dynamodb.write_versioned as wv
+
+# construct these at a module level and share them - they are
+# threadsafe because they are pure values.
+task_table = wv.ItemTable('TaskTable', item_name='Task')
+user_table = wv.ItemTable('UserTable', item_name='User')
+
+# imagine these were passed in via an API
+task_key = dict(id="task1")
+new_task = dict(task_key, name="plant tree", num_subtasks=3, user_id="steve")
+
+# this transaction builder is a pure function that constructs
+# a value representing DynamoDB effects to be transactionally/atomically applied.
+def create_and_link_task_unless_exists(t: wv.VersionedTransaction) -> wv.VersionedTransaction:
+    t = task_table.presume(task_key, None)(t)
+    # ^ we presume that the task does not exist, i.e. has the value None
+    # This is a no-op if a value has already been fetched.
+    # This is also purely an optimization to avoid an initial read;
+    # the whole transaction would result in the same data in the table without it.
+
+    existing_task = task_table.get(task_key)(t)
+    if existing_task:
+        return t
+        # ^ perform no action as long as the item already exists
+
+    t = task_table.put(new_task)(t)
+    # ^ put this item into this table as long as the item does not exist
+    user = user_table.require(dict(id=new_task["user_id"]))(t)
+    # ^ fetch the user and raise an exception if it does not exist
+    user["task_ids"].append(task_key["id"])
+    t = user_table.put(user)(t)
+    # ^ make sure that the user knows about its task
+    return t
+    # ^ return the built transaction to be executed.
+
+# this call will cause the transaction builder function above
+# to be called one to many times until its logic is atomically
+# applied (there are no intervening writes to any of the items),
+# or until a (configurable-with-default) timeout is reached.
+wv.versioned_transact_write_items(
+    create_and_link_task_unless_exists,
+    attempts_iterator=wv.timed_retry(timedelta(seconds=22.3)
+)
+```
+
+### Prefetching items to transact
+
+In many cases you will wish you prefetch some items that you do expect
+to exist, since you plan to update them. Items that have not been
+prefetched will be fetched at the time of the call to `wv.get` or
+`wv.require` in a running transaction, but these fetches will be done
+serially, so it is more efficient to provide their keys
+upfront. `wv.put` and `wv.delete` will not perform a
+read-before-write, but will instead optimistically attempt the write,
+assuming that the item does not exist until proven otherwise.
+
+To perform those prefetches, simply provide a set of keys mapped onto
+by table names, as so:
+
+```python
+wv.versioned_transact_write_items(
+    transfer_task_ownership,
+    dict(
+        TaskTable=[task_key1, task_key2],
+        UserTable=[current_owner_key, new_owner_key],
+    )
+)
+```
+
+### Unit testing!
+
+A big advantage to this approach, besides the readability of the code
+vs the standard `boto3` APIs, and the simplicity of knowing that you're
+never risking data loss with less sophisticated writes to your table,
+is that all of your business logic can be expressed directly in terms
+of interacting with the database, but you can fully unit test that behavior.
+
+See the
+[included example unit tests](../../../tests/xoto3/dynamodb/write_versioned/example_unit_test.py)
+for a working example of unit testing the above
+`create_and_link_task_unless_exists` function.
diff --git a/xoto3/dynamodb/write_versioned/__init__.py b/xoto3/dynamodb/write_versioned/__init__.py
index e0f2023..d373a59 100644
--- a/xoto3/dynamodb/write_versioned/__init__.py
+++ b/xoto3/dynamodb/write_versioned/__init__.py
@@ -1,11 +1,24 @@
-"""API for transact_write_items"""
-from .errors import TableSchemaUnknownError, TransactionAttemptsOverrun  # noqa
+"""API for transact_write_items.
+
+Anything you import from any underlying module directly (i.e. not from
+here) is not guaranteed to remain, so don't do that. Import this
+module and use only what it exposes.
+"""
+from .api2 import ItemTable, TypedTable, create_or_update, update_existing, update_if_exists  # noqa
+from .errors import (  # noqa
+    ItemUndefinedException,
+    TableSchemaUnknownError,
+    TransactionAttemptsOverrun,
+)
 from .modify import delete, put  # noqa
 from .read import get, require  # noqa
+from .retry import timed_retry  # noqa
 from .run import versioned_transact_write_items  # noqa
+from .specify import define_table, presume  # noqa
 from .types import (  # noqa
     ItemKeysByTableName,
     ItemsByTableName,
+    TransactionBuilder,
     TransactWriteItems,
     VersionedTransaction,
 )
diff --git a/xoto3/dynamodb/write_versioned/api2.py b/xoto3/dynamodb/write_versioned/api2.py
new file mode 100644
index 0000000..5f2ad25
--- /dev/null
+++ b/xoto3/dynamodb/write_versioned/api2.py
@@ -0,0 +1,170 @@
+"""
+A table and type-oriented API for using versioned_transact_write_items.
+
+All of these things are simple bolt-ons for the underlying
+functionality, and may safely be ignored.
+"""
+from copy import deepcopy
+from typing import Callable, Generic, Optional, TypeVar, Union
+
+from .ddb_api import table_name
+from .modify import delete, put
+from .read import get, require
+from .specify import define_table, presume
+from .types import Item, ItemKey, TableNameOrResource, TransactionBuilder, VersionedTransaction
+
+T = TypeVar("T")
+
+Thunk = Callable[[], T]
+# a protocol for a simplified form of Lazy, a.k.a. a Thunk - the thing
+# that you do when you want to defer realizing a value until runtime.
+
+
+class TypedTable(Generic[T]):
+    """This is a table-centric API wrapper for the low-level transaction
+    interaction capabilities provided by `read.py` and
+    `modify.py`. There are simplifed docs here, and you can reference
+    the underlying implementations for more comprehensive
+    documentation.
+
+    In all cases, these methods are ways of interacting with an opaque
+    VersionedTransaction object which represents the state of the
+    database before the transaction began
+
+    In practice, it's nice to have table-aware closures of these
+    functions when dealing with a table, versus providing the string
+    table name to every call. This is especially valuable if you're
+    dealing with only a single table over and over again, and there's
+    some kind of de/serialization you want to run consistently against
+    that table.
+
+    Note that your type_deserializer is now the expected method of
+    performing the standard deepcopy-on-get to prevent mutating the
+    immutable items retrieved from a VersionedTransaction.  Therefore
+    the standard copy-on-get will not be performed. If you have no
+    need for additional type deserialization, you must provide
+    `copy.deepcopy`, or use `ItemTable`, to retain the standard, safe
+    behavior.
+
+    """
+
+    def __init__(
+        self,
+        table_name: Union[TableNameOrResource, Thunk[TableNameOrResource]],
+        type_deserializer: Callable[[Item], T],
+        # from the database to your code - on the read path
+        type_serializer: Callable[[T], Item],
+        # from your code to the database - on the write path
+        item_name: str = "Item",
+    ):
+        """Construct a table-centric API utility for a given table with
+        shared serialization and deserialization.
+
+        Your type deserializer must implement a form of deep copy.
+        """
+        if not callable(table_name):
+            self.lazy_table = lambda: table_name
+        else:
+            self.lazy_table = table_name
+        self.item_name = item_name
+        self.type_deserializer = type_deserializer
+        self.type_serializer = type_serializer
+
+    def _lazy_table_name(self) -> str:
+        return table_name(self.lazy_table())
+
+    def get(self, key: ItemKey) -> Callable[[VersionedTransaction], Optional[T]]:
+        """Get an item from the database if it exists"""
+
+        def deser_opt(item: Optional[Item]) -> Optional[T]:
+            return self.type_deserializer(item) if item else None
+
+        return lambda vt: deser_opt(
+            get(vt, self._lazy_table_name(), key, copy=False, nicename=self.item_name)
+        )
+
+    def require(self, key: ItemKey) -> Callable[[VersionedTransaction], T]:
+        """Return the item for this key, or raise an ItemNotFoundException if it does not"""
+        return lambda vt: self.type_deserializer(
+            require(vt, self._lazy_table_name(), key, copy=False, nicename=self.item_name)
+        )
+
+    def put(self, typed_item: T) -> TransactionBuilder:
+        return lambda vt: put(
+            vt, self._lazy_table_name(), self.type_serializer(typed_item), nicename=self.item_name
+        )
+
+    def delete(self, key: ItemKey) -> TransactionBuilder:
+        return lambda vt: delete(vt, self._lazy_table_name(), key, nicename=self.item_name)
+
+    def presume(self, key: ItemKey, value: Optional[T]) -> TransactionBuilder:
+        """'To assume as true in the absence of proof to the contrary.'
+
+        Returns a modified transaction with this value set if the value of
+        the item is not already known. If a value has already been fetched
+        or presumed, this will be a no-op.
+
+        See further docs in .read.py.
+        """
+        return lambda vt: presume(
+            vt,
+            self._lazy_table_name(),
+            key,
+            None if value is None else self.type_serializer(value),
+        )
+
+    def define(self, *key_attributes: str) -> TransactionBuilder:
+        """Idempotent definition of key attributes for a table without any I/O"""
+        return lambda vt: define_table(vt, self._lazy_table_name(), *key_attributes)
+
+
+def ItemTable(
+    table_name: Union[Thunk[TableNameOrResource], TableNameOrResource], item_name: str = "Item"
+) -> TypedTable[Item]:
+    """This is just a workaround for the fact that mypy can't handle
+    generics with default arguments, e.g. in the TypedTable
+    constructor above.
+    """
+
+    def _item_ident(item: Item) -> Item:
+        return item
+
+    return TypedTable(table_name, deepcopy, _item_ident, item_name=item_name)
+
+
+def update_if_exists(
+    table: TypedTable[T], updater: Callable[[T], T], key: ItemKey,
+) -> TransactionBuilder:
+    """Does not call the updater function if the item does not exist in the table."""
+
+    def _update_if_exists(vt: VersionedTransaction) -> VersionedTransaction:
+        item = table.get(key)(vt)
+        if item:
+            return table.put(updater(item))(vt)
+        return vt
+
+    return _update_if_exists
+
+
+def update_existing(
+    table: TypedTable[T], updater: Callable[[T], T], key: ItemKey,
+) -> TransactionBuilder:
+    """Raises ItemNotFoundException if the item to be updated does not exist in the table."""
+
+    def update_translator(vt: VersionedTransaction) -> VersionedTransaction:
+        return table.put(updater(table.require(key)(vt)))(vt)
+
+    return update_translator
+
+
+def create_or_update(
+    table: TypedTable[T], creator_updater: Callable[[Optional[T]], T], key: ItemKey,
+) -> TransactionBuilder:
+    """Provides the item if it exists, or None if it does not, but expects
+    your callable to return a writeable item.
+    """
+
+    def create_or_update_trans(vt: VersionedTransaction) -> VersionedTransaction:
+        return table.put(creator_updater(table.get(key)(vt)))(vt)
+
+    return create_or_update_trans
diff --git a/xoto3/dynamodb/write_versioned/ddb_api.py b/xoto3/dynamodb/write_versioned/ddb_api.py
index a053e02..4c8a934 100644
--- a/xoto3/dynamodb/write_versioned/ddb_api.py
+++ b/xoto3/dynamodb/write_versioned/ddb_api.py
@@ -4,7 +4,6 @@
 from logging import getLogger
 from typing import Any, Dict, List, Mapping, Optional, Set, Tuple, cast
 
-import boto3
 from botocore.exceptions import ClientError
 from typing_extensions import Protocol, TypedDict
 
@@ -12,6 +11,7 @@
 from xoto3.dynamodb.utils.serde import serialize_item
 from xoto3.dynamodb.utils.table import table_primary_keys
 from xoto3.errors import client_error_name
+from xoto3.lazy_session import tll_from_session
 
 from .keys import hashable_key_to_key
 from .types import (
@@ -34,6 +34,10 @@
 }
 
 
+_DDB_RES = tll_from_session(lambda s: s.resource("dynamodb"))
+_DDB_CLIENT = tll_from_session(lambda s: s.client("dynamodb"))
+
+
 def table_name(table: TableNameOrResource) -> str:
     if not isinstance(table, str):
         return table.name
@@ -41,9 +45,25 @@ def table_name(table: TableNameOrResource) -> str:
 
 
 def known_key_schema(table: TableNameOrResource) -> Tuple[str, ...]:
+    """Note that this code may perform I/O, and may run during the
+    operation of your transaction builder.
+
+    Technically that is a side effect, and also this requires
+    permissions to read the key schema of your table using
+    DynamoDB::DescribeTable.
+
+    You can avoid ever having this happen just by making sure to
+    define the key schema of your table prior to performing a put or
+    delete, either via a get/require, or via the define_table
+    mechanism. In the vast majority of cases, you should not find
+    yourself in this situation, but if you do get an exception
+    bubbling up through here, that's what's going on, and you should
+    probably add a call to `define_table` at the beginning of your
+    transaction.
+    """
     try:
         if isinstance(table, str):
-            table = boto3.resource("dynamodb").Table(table)
+            table = _DDB_RES().Table(table)
         assert not isinstance(table, str)
         # your environment may or may not have permissions to read the key schema of its table.
         # in general, that is a nice permission to allow if possible.
@@ -82,8 +102,9 @@ def _ddb_batch_get_item(
         for table_name, item_keys in item_keys_by_table_name.items()
         if item_keys  # don't make empty request to a table
     }
-    results = defaultdict(list)
+    results: Dict[str, List[Item]] = defaultdict(list)
     while unprocessed_keys:
+        _log.debug(f"Performing batch_get of {len(unprocessed_keys)} keys")
         response = batch_get_item(RequestItems=unprocessed_keys)
         unprocessed_keys = response.get("UnprocessedKeys")  # type: ignore
         for table_name, items in response["Responses"].items():
@@ -113,14 +134,12 @@ def boto3_impl_defaults(
     batch_get_item: Optional[BatchGetItem] = None,
     transact_write_items: Optional[TransactWriteItems] = None,
 ) -> Tuple[BatchGetItem, TransactWriteItems]:
-    resource = None
     if not batch_get_item:
-        resource = boto3.resource("dynamodb")
-        batch_get_item = cast(BatchGetItem, partial(_ddb_batch_get_item, resource.batch_get_item))
+        batch_get_item = cast(BatchGetItem, partial(_ddb_batch_get_item, _DDB_RES().batch_get_item))
     assert batch_get_item
 
     if not transact_write_items:
-        transact_write_items = make_transact_multiple_but_optimize_single(boto3.client("dynamodb"))
+        transact_write_items = make_transact_multiple_but_optimize_single(_DDB_CLIENT())
     assert transact_write_items
 
     return (
@@ -233,6 +252,7 @@ def put_or_delete_item(item_hashable_key: HashableItemKey, effect: Optional[Item
         def item_remains_unmodified(
             item_hashable_key: HashableItemKey, item: Optional[Item]
         ) -> dict:
+            """This will also check that the item still does not exist if it previously did not"""
             expression_ensuring_unmodifiedness = _serialize_versioned_expr(
                 versioned_item_expression(
                     get_existing_item(item_hashable_key).get(item_version_attribute, 0),
diff --git a/xoto3/dynamodb/write_versioned/errors.py b/xoto3/dynamodb/write_versioned/errors.py
index 6f099bf..9ceb2fc 100644
--- a/xoto3/dynamodb/write_versioned/errors.py
+++ b/xoto3/dynamodb/write_versioned/errors.py
@@ -22,10 +22,15 @@ class TableSchemaUnknownError(KeyError):
     """
 
 
-class ItemNotYetFetchedError(DynamoDbItemException):
+class ItemUndefinedException(DynamoDbItemException):
     """Used internally to send control flow and information back to the
-    transaction runner to prompt a lazy load of an item not already
-    fetched.
+    transaction runner to prompt a lazy load of an item whose state in
+    the table is not currently known because a fetch has never been
+    attempted.
+
+    You should never see one of these errors unless you're writing unit tests
+    and didn't define an initial value for your items with `presume`.
+
     """
 
 
diff --git a/xoto3/dynamodb/write_versioned/keys.py b/xoto3/dynamodb/write_versioned/keys.py
index 27aab94..654a7db 100644
--- a/xoto3/dynamodb/write_versioned/keys.py
+++ b/xoto3/dynamodb/write_versioned/keys.py
@@ -1,6 +1,6 @@
 """Private implementation details for versioned_transact_write_items"""
 
-from typing import Collection, Sequence, Union
+from typing import Collection, Sequence, Tuple, Union
 
 from xoto3.dynamodb.types import Item, ItemKey
 
@@ -30,3 +30,7 @@ def hashable_key_to_key(key_attributes: Sequence[str], hashable_key: HashableIte
         return {key_attributes[0]: hashable_key[0], key_attributes[1]: hashable_key[1]}
     assert len(key_attributes) == 1
     return {key_attributes[0]: hashable_key}
+
+
+def standard_key_attributes(*key_attrs: str) -> Tuple[str, ...]:
+    return tuple(sorted(key_attrs))
diff --git a/xoto3/dynamodb/write_versioned/lazy_batch_gets.py b/xoto3/dynamodb/write_versioned/lazy_batch_gets.py
new file mode 100644
index 0000000..b323081
--- /dev/null
+++ b/xoto3/dynamodb/write_versioned/lazy_batch_gets.py
@@ -0,0 +1,56 @@
+from typing import Collection, Mapping, Tuple
+
+from .errors import ItemUndefinedException
+from .prepare import add_item_to_base_request, parse_batch_get_request, prepare_clean_transaction
+from .specify import presume
+from .types import BatchGetItem, ItemKey, TransactionBuilder, VersionedTransaction
+
+
+def lazy_batch_getting_transaction_builder(
+    transaction_builder: TransactionBuilder,
+    item_keys_by_table_name: Mapping[str, Collection[ItemKey]],
+    batch_get_item: BatchGetItem,
+) -> Tuple[VersionedTransaction, VersionedTransaction]:
+    """If your transaction attempts to get/require an item that was not
+    prefetched, we will stop, fetch it, and then retry your
+    transaction from the beginning.
+
+    Remember to make your transaction a pure function, or at least make
+    sure all of its side effects are effectively idempotent!
+    """
+    # this outer loop lets us repeat the batch_get logic for actually
+    # fetching needed items from the DynamoDB tables.
+    while True:
+        item_keys_by_table_name = parse_batch_get_request(item_keys_by_table_name)
+        clean_transaction = prepare_clean_transaction(
+            item_keys_by_table_name, batch_get_item(item_keys_by_table_name),
+        )
+        undefined_needing_fetch = False
+        # the goal of this loop is to accumulate as many lazy `get`s
+        # as possible before actually executing them above.
+        #
+        # There isn't a general purpose way to 'know' about all the
+        # gets that you might end up wanting to do, but if you have a
+        # series of `get`s uninterrupted by `require`s, they will
+        # function as a batch get.
+        while True:
+            try:
+                built_transaction = transaction_builder(clean_transaction)
+                if not undefined_needing_fetch:
+                    # the builder didn't need to ask for anything
+                    return clean_transaction, built_transaction
+                else:
+                    # the builder has asked for everything it knows to ask for
+                    # and we need to restart from the batch get above.
+                    break
+            except ItemUndefinedException as inf:
+                assert inf.key
+                item_keys_by_table_name = add_item_to_base_request(
+                    item_keys_by_table_name, (inf.table_name, inf.key),
+                )
+                if getattr(inf, "force_immediate_fetch", False):
+                    # perform immediate fetch; without it the transaction cannot proceed
+                    break
+                undefined_needing_fetch = True
+                # presume that the value is None for now, to let this `get` pass instead of excepting.
+                clean_transaction = presume(clean_transaction, inf.table_name, inf.key, None)
diff --git a/xoto3/dynamodb/write_versioned/modify.py b/xoto3/dynamodb/write_versioned/modify.py
index 916ca50..a4eb266 100644
--- a/xoto3/dynamodb/write_versioned/modify.py
+++ b/xoto3/dynamodb/write_versioned/modify.py
@@ -1,4 +1,4 @@
-"""Defines the API for modifying an existing transaction. """
+"""Defines the API for adding write effects to a transaction."""
 
 from typing import NamedTuple, Optional, Union
 
@@ -10,8 +10,7 @@
 from .ddb_api import known_key_schema
 from .ddb_api import table_name as _table_name
 from .errors import TableSchemaUnknownError
-from .keys import hashable_key, key_from_item
-from .prepare import standard_key_attributes_from_key
+from .keys import hashable_key, key_from_item, standard_key_attributes
 from .types import TableNameOrResource, VersionedTransaction, _TableData
 
 
@@ -33,14 +32,7 @@ def _write(
     *,
     nicename: str = DEFAULT_ITEM_NAME,
 ) -> VersionedTransaction:
-    """Shared put/delete implementation - not meant for direct use at this time.
-
-    Performs an optimistic put - if the item is not known to the
-    existing transaction, assumes you mean to create it if and only if
-    it does not exist. If the item turns out to exist already, your
-    transaction will be re-run, at which point a put will be interpreted as a
-    'witting' choice to overwrite the known item.
-    """
+    """Shared put/delete implementation - not meant for direct use at this time."""
 
     table_name = _table_name(table)
 
@@ -63,7 +55,7 @@ def _write(
                         "Specify this delete in terms of the key only and this should work fine."
                     )
                 # at this point this is a best guess
-                key_attributes = standard_key_attributes_from_key(put_or_delete.item_key)
+                key_attributes = standard_key_attributes(*put_or_delete.item_key.keys())
             else:
                 # it's a put - we can't make this work at all
                 raise TableSchemaUnknownError(
@@ -100,7 +92,14 @@ def put(
     nicename: str = DEFAULT_ITEM_NAME,
     prewrite_transform: Optional[SimpleTransform] = None,
 ) -> VersionedTransaction:
-    """Returns a modified transaction including the requested PutItem operation"""
+    """Returns a modified transaction including the requested PutItem operation
+
+    Performs an optimistic put - if the item is not known to the
+    existing transaction, assumes you mean to create it if and only if
+    it does not exist. If the item turns out to exist already, your
+    transaction will be re-run, at which point a put will be interpreted as a
+    'witting' choice to overwrite the known item.
+    """
     return _write(
         transaction, table, Put(dynamodb_prewrite(item, prewrite_transform)), nicename=nicename
     )
diff --git a/xoto3/dynamodb/write_versioned/prepare.py b/xoto3/dynamodb/write_versioned/prepare.py
index 013d114..2e4261d 100644
--- a/xoto3/dynamodb/write_versioned/prepare.py
+++ b/xoto3/dynamodb/write_versioned/prepare.py
@@ -2,7 +2,7 @@
 
 from xoto3.dynamodb.types import Item, ItemKey, KeyAttributeType
 
-from .keys import hashable_key, hashable_key_to_key, key_from_item
+from .keys import hashable_key, hashable_key_to_key, key_from_item, standard_key_attributes
 from .types import HashableItemKey, VersionedTransaction, _TableData
 
 
@@ -10,7 +10,7 @@ def _deduplicate_and_validate_keys(keys: Collection[ItemKey]) -> Iterable[ItemKe
     """All keys for a given table will need to be unique, and of course
     they must all share the same attribute names (or else they do not
     match the key schema for the table."""
-    seen = set()
+    seen: Set[tuple] = set()
     key_attributes: Set[KeyAttributeType] = set()
     for key in keys:
         tk = tuple(key.items())
@@ -56,13 +56,9 @@ def items_and_keys_to_clean_table_data(
     )
 
 
-def standard_key_attributes_from_key(item_key: ItemKey) -> Tuple[str, ...]:
-    return tuple(sorted(item_key.keys()))
-
-
 def _extract_key_attributes(item_keys: Sequence[ItemKey]) -> Tuple[str, ...]:
     assert item_keys, "You can't extract key_attributes without at least one item_key."
-    return standard_key_attributes_from_key(item_keys[0])
+    return standard_key_attributes(*item_keys[0].keys())
 
 
 D = TypeVar("D", bound=dict)
diff --git a/xoto3/dynamodb/write_versioned/read.py b/xoto3/dynamodb/write_versioned/read.py
index f9bcad3..0287198 100644
--- a/xoto3/dynamodb/write_versioned/read.py
+++ b/xoto3/dynamodb/write_versioned/read.py
@@ -8,7 +8,7 @@
 from xoto3.dynamodb.types import Item, ItemKey
 
 from .ddb_api import table_name as _table_name
-from .errors import ItemNotYetFetchedError
+from .errors import ItemUndefinedException
 from .keys import hashable_key
 from .types import TableNameOrResource, VersionedTransaction
 
@@ -27,6 +27,10 @@ def get(
 ) -> Optional[Item]:
     """Returns current state of the transactable item according to the
     transaction.
+
+    If we have no knowledge of the item, we will restart your
+    transaction after attempting a fetch of this item.
+
     This is Python, so the only way we can stop you from modifying
     this canonical reference for the current value of the item is to
     return a deep copy. That's expensive, so if you trust yourself you
@@ -34,11 +38,12 @@ def get(
     HOWEVER, the behavior of this system is **undefined** if you
     disable this behavior and then modify one of the retrieved items
     directly. Caveat emptor...
+
     """
     table_name = _table_name(table)
-    item_exc_type = get_item_exception_type(nicename, ItemNotYetFetchedError)
+    UndefinedException = get_item_exception_type(nicename, ItemUndefinedException)
     if table_name not in transaction.tables:
-        raise item_exc_type("Table new to transaction", key=item_key, table_name=table_name)
+        raise UndefinedException("Table new to transaction", key=item_key, table_name=table_name)
 
     items, effects, _ = transaction.tables[table_name]
     item_hashable_key = hashable_key(item_key)
@@ -47,7 +52,7 @@ def get(
     if item_hashable_key in effects:
         return xf_result(effects[item_hashable_key])
     if item_hashable_key not in items:
-        raise item_exc_type(
+        raise UndefinedException(
             f"{nicename} not yet present in transaction", key=item_key, table_name=table_name
         )
 
@@ -62,11 +67,19 @@ def require(
     nicename: str = DEFAULT_ITEM_NAME,
     **kwargs,
 ) -> Item:
+    """Returns the current state of the transactable item according to the
+    transaction, or raises an ItemNotFoundException if it does not
+    exist in the table.
+    """
     table_name = _table_name(table)
-    item = get(transaction, table_name, item_key, nicename=nicename, **kwargs)
-    if not item:
-        raise_if_empty_getitem_response(
-            dict(), nicename=nicename, key=item_key, table_name=table_name
-        )
-    assert item is not None
-    return item
+    try:
+        item = get(transaction, table_name, item_key, nicename=nicename, **kwargs)
+        if not item:
+            raise_if_empty_getitem_response(
+                dict(), nicename=nicename, key=item_key, table_name=table_name
+            )
+        assert item is not None
+        return item
+    except ItemUndefinedException as inf:
+        inf.force_immediate_fetch = True  # type: ignore
+        raise inf
diff --git a/xoto3/dynamodb/write_versioned/retry.py b/xoto3/dynamodb/write_versioned/retry.py
index f9facc0..42e2947 100644
--- a/xoto3/dynamodb/write_versioned/retry.py
+++ b/xoto3/dynamodb/write_versioned/retry.py
@@ -1,5 +1,6 @@
 import random
 import time
+from datetime import timedelta
 from logging import getLogger
 from typing import Iterator
 
@@ -25,12 +26,12 @@ def _choose_sleep_len_to_average_N_attempts_in_the_total_interval(
 
 
 def timed_retry(
-    transaction_expiration_seconds: float = 5.0,
+    transaction_expiration: timedelta = timedelta(seconds=5.0),
     max_attempts_before_expiration: int = 25,
     random_sleep_length: bool = True,
 ) -> Iterator:
     attempt = 0
-    expiring_at = time.monotonic() + transaction_expiration_seconds
+    expiring_at = time.monotonic() + transaction_expiration.total_seconds()
 
     while attempt == 0 or time.monotonic() <= expiring_at:
         attempt += 1
diff --git a/xoto3/dynamodb/write_versioned/run.py b/xoto3/dynamodb/write_versioned/run.py
index 99c187d..da12c53 100644
--- a/xoto3/dynamodb/write_versioned/run.py
+++ b/xoto3/dynamodb/write_versioned/run.py
@@ -1,7 +1,7 @@
 """The core run loop for a transaction"""
 from datetime import datetime
 from logging import getLogger
-from typing import Callable, Collection, Iterable, Mapping, Optional, Tuple
+from typing import Callable, Collection, Iterable, Mapping, Optional
 
 from botocore.exceptions import ClientError
 
@@ -13,13 +13,9 @@
     built_transaction_to_transact_write_items_args,
     is_cancelled_and_retryable,
 )
-from .errors import ItemNotYetFetchedError, TransactionAttemptsOverrun
-from .prepare import (
-    add_item_to_base_request,
-    all_items_for_next_attempt,
-    parse_batch_get_request,
-    prepare_clean_transaction,
-)
+from .errors import TransactionAttemptsOverrun
+from .lazy_batch_gets import lazy_batch_getting_transaction_builder
+from .prepare import all_items_for_next_attempt
 from .retry import timed_retry
 from .types import BatchGetItem, TransactionBuilder, TransactWriteItems, VersionedTransaction
 
@@ -30,38 +26,10 @@ def _is_empty(transaction: VersionedTransaction) -> bool:
     return not sum(len(effects) for _, effects, *_ in transaction.tables.values())
 
 
-def _lazy_loading_transaction_builder(
-    transaction_builder: TransactionBuilder,
-    item_keys_by_table_name: Mapping[str, Collection[ItemKey]],
-    batch_get_item: BatchGetItem,
-) -> Tuple[VersionedTransaction, VersionedTransaction]:
-    """If your transaction attempts to get/require an item that was not
-    prefetched, we will stop, fetch it, and then retry your
-    transaction from the beginning.
-
-    Remember to make your transaction a pure function, or at least make
-    sure all of its side effects are effectively idempotent!
-    """
-    while True:
-        item_keys_by_table_name = parse_batch_get_request(item_keys_by_table_name)
-        clean_transaction = prepare_clean_transaction(
-            item_keys_by_table_name, batch_get_item(item_keys_by_table_name),
-        )
-        try:
-            built_transaction = transaction_builder(clean_transaction)
-            # the built transaction will always contain a superset of the keys
-            # present in the clean transaction.
-            return clean_transaction, built_transaction
-        except ItemNotYetFetchedError as item_error:
-            assert item_error.key
-            item_keys_by_table_name = add_item_to_base_request(
-                item_keys_by_table_name, (item_error.table_name, item_error.key),
-            )
-
-
 def versioned_transact_write_items(
     transaction_builder: TransactionBuilder,
     item_keys_by_table_name: Mapping[str, Collection[ItemKey]] = dict(),
+    # specify any items you want prefetched
     *,
     batch_get_item: Optional[BatchGetItem] = None,
     transact_write_items: Optional[TransactWriteItems] = None,
@@ -70,16 +38,22 @@ def versioned_transact_write_items(
     item_version_attribute: str = "item_version",
     last_written_attribute: str = "last_written_at",
 ) -> VersionedTransaction:
-    """Performs a read-transact-write loop over multiple items across
-    multiple tables until there are no interruptions causing the
-    transaction to fail, or until the transaction expires (as
-    specified by the attempts_iterator or the default timed_retry).
-
-    Allows PUTs and DELETEs - an UPDATE with versioning is no different
-    from a PUT.
-
-    You do not have to write back to every item specified as part of
-    the transaction, but we will assert the unchanged state of every
+    """Performs a read-transact-write loop over multiple versioned
+    puts/deletes across multiple tables, until there are no
+    interruptions causing the transaction to fail, or until the
+    transaction expires (as specified by the attempts_iterator or the
+    default timed_retry).
+
+    Your transaction _must_ be a pure function, as it will likely be
+    called more than once while attempting to resolve all your data
+    dependencies as specified by `get/require`, and may also need to
+    be retried because of competing transactions.
+
+    Allows only PUTs and DELETEs - an UPDATE with versioning is no
+    different from a PUT.
+
+    You do not have to write back to every item fetched for the
+    transaction, but we will assert the unchanged state of every
     non-modified object as part of the transaction. This allows
     building of transactions where the micro-verse of your transaction
     is known to be the fully consistent state as returned at the
@@ -87,16 +61,16 @@ def versioned_transact_write_items(
     you did not need to modify.
 
     Returns the completed transaction, which contains the resulting
-    items as written to the table(s).
+    items as written to the table(s) at the completion of the
+    transaction.
 
     In the future we could add support for narrower ConditionChecks on
     unwritten items, rather than the present general assertion that
     their `item_version` was not changed.
 
-    The default implementation for transact_write_items will also
-    attempt to optimize your usage by reverting to a simple Put or
-    Delete if you only operate on a single item.
-
+    The implementation for transact_write_items will also optimize
+    your DynamoDB costs by reverting to a simple (but still versioned)
+    Put or Delete if you only operate on a single item.
     """
     batch_get_item, transact_write_items = boto3_impl_defaults(
         batch_get_item, transact_write_items,
@@ -104,7 +78,7 @@ def versioned_transact_write_items(
 
     built_transaction = None
     for i, _ in enumerate(attempts_iterator or timed_retry()):
-        clean_transaction, built_transaction = _lazy_loading_transaction_builder(
+        clean_transaction, built_transaction = lazy_batch_getting_transaction_builder(
             transaction_builder, item_keys_by_table_name, batch_get_item,
         )
         if built_transaction is clean_transaction or _is_empty(built_transaction):
diff --git a/xoto3/dynamodb/write_versioned/specify.py b/xoto3/dynamodb/write_versioned/specify.py
new file mode 100644
index 0000000..77bc6da
--- /dev/null
+++ b/xoto3/dynamodb/write_versioned/specify.py
@@ -0,0 +1,96 @@
+"""Low level API for specifying non-effect facts about the state of the database"""
+from typing import Optional
+
+from .ddb_api import table_name as _table_name
+from .keys import hashable_key, standard_key_attributes
+from .types import Item, ItemKey, TableNameOrResource, VersionedTransaction, _TableData
+
+
+def presume(
+    transaction: VersionedTransaction,
+    table: TableNameOrResource,
+    item_key: ItemKey,
+    item_value: Optional[Item],
+) -> VersionedTransaction:
+
+    """'To assume as true in the absence of proof to the contrary.'
+
+    Returns a modified transaction with this value set if the value of
+    the item is not already known. If a value has already been fetched
+    or presumed, this will be a no-op.
+
+    If modified, the presumed value will be available via `get`, and
+    will additionally check your presumed value against the table when
+    the transaction is run.
+
+    At runtime (within the context of
+    `versioned_transact_write_items`), this is purely a cost
+    optimization to avoid fetching an item for which you believe you
+    already know the value. Whether your presumption is right, wrong,
+    or you don't presume anything, your transaction builder will
+    result in exactly the same data written to the table.
+
+    As with any item fetched or presumed, if the item turns out to
+    have a different value in the table than you presumed when the
+    transaction is committed, the transaction will restart, the item
+    will be freshly fetched like usual, and your use of presumption
+    will have no ultimate effect on the data.
+
+    For unit testing, this is the approved approach for setting up a
+    VersionedTransaction 'fixture', where you declare the state of the
+    database before your transaction builder is run. Set a presumed
+    value for every item that you will attempt to `get` or `require`
+    within your transaction builder, otherwise your test will error
+    with ItemUndefinedException.
+
+    """
+    if item_value is not None:
+        for key_attr, key_val in item_key.items():
+            assert item_value[key_attr] == key_val, "Item key must match in a non-nil item value"
+    table_name = _table_name(table)
+    if table_name in transaction.tables:
+        table_data = transaction.tables[table_name]
+    else:
+        table_data = _TableData(
+            items=dict(), effects=dict(), key_attributes=standard_key_attributes(*item_key.keys())
+        )
+    hkey = hashable_key(item_key)
+    if hkey not in table_data.items:
+        return VersionedTransaction(
+            tables={
+                **transaction.tables,
+                table_name: _TableData(
+                    items={**table_data.items, hkey: item_value},
+                    effects=table_data.effects,
+                    key_attributes=table_data.key_attributes,
+                ),
+            }
+        )
+    return transaction
+
+
+def define_table(
+    transaction: VersionedTransaction, table: TableNameOrResource, *key_attributes: str,
+) -> VersionedTransaction:
+    """Idempotent definition of key attribute schema for the given table
+    without forcing any IO operations/effects up front.
+
+    The main reason you might want to do this is if you need to do a
+    `put`, because `put` cannot infer the shape of your key.
+
+    If the table definition is already present, this is a no-op.
+
+    """
+    assert len(key_attributes) > 0 and len(key_attributes) <= 2
+    if _table_name(table) in transaction.tables:
+        return transaction
+    return VersionedTransaction(
+        tables={
+            **transaction.tables,
+            _table_name(table): _TableData(
+                items=dict(),
+                effects=dict(),
+                key_attributes=standard_key_attributes(*key_attributes),
+            ),
+        }
+    )
diff --git a/xoto3/dynamodb/write_versioned/types.py b/xoto3/dynamodb/write_versioned/types.py
index e9fe8e7..936de96 100644
--- a/xoto3/dynamodb/write_versioned/types.py
+++ b/xoto3/dynamodb/write_versioned/types.py
@@ -30,10 +30,15 @@ class _TableData(NamedTuple):
 
 
 class VersionedTransaction(NamedTuple):
-    """A container suitable for functional usage of get, put, and delete.
+    """An opaque object representing the state of the database at the time
+    the transaction began, plus all effects defined by modifier
+    operations (put, delete), plus database specifications defined by
+    the presume and define_table operations.
+
+    The internals of this type does NOT constitute an API, and you
+    should never modify this object except via the provided API
+    functions.
 
-    This type does NOT constitute an API, and you should never modify
-    this object directly.
     """
 
     tables: Mapping[str, _TableData]
diff --git a/xoto3/lazy_session.py b/xoto3/lazy_session.py
index 28f0249..9192d84 100644
--- a/xoto3/lazy_session.py
+++ b/xoto3/lazy_session.py
@@ -1,15 +1,18 @@
-import boto3
+"""Apparently there are thread-safety issues conditions if you don't
+first create a session.
 
-from xoto3.utils.lazy import ThreadLocalLazy
+https://boto3.amazonaws.com/v1/documentation/api/latest/guide/session.html#multithreading-or-multiprocessing-with-sessions
 
+https://github.com/boto/boto3/issues/1592
+"""
+from typing import Callable, TypeVar
 
-class SessionedBoto3:
-    """Apparently there are thread-safety issues conditions if you don't
-    first create a session.
+import boto3.session
 
-    https://github.com/boto/boto3/issues/1592
-    """
+from xoto3.utils.lazy import ThreadLocalLazy
 
+
+class SessionedBoto3:
     def __init__(self, method_name: str, *args, **kwargs):
         self.method_name = method_name
         assert self.method_name in {"resource", "client"}
@@ -23,5 +26,29 @@ def __call__(self):
 
 
 def tlls(method_name: str, *args, **kwargs) -> ThreadLocalLazy[SessionedBoto3]:
-    """Thread Local Lazy SessionedBoto3"""
+    """Thread Local Lazy SessionedBoto3.
+
+    This is deprecated in favor of tll_from_session because this one
+    doesn't allow any sort of static typing to pass through.
+    """
     return ThreadLocalLazy(SessionedBoto3(method_name, *args, **kwargs))
+
+
+RC = TypeVar("RC")
+
+
+_THREAD_SESSION = ThreadLocalLazy(lambda: boto3.session.Session())
+# instead of using `resource` and `client` directly, it's recommended
+# that you use this per-thread session to then create a ThreadLocalLazy resource or client.
+
+
+def tll_from_session(
+    resource_or_client_creator: Callable[[boto3.session.Session], RC]
+) -> ThreadLocalLazy[RC]:
+    """Gives your callback a threadsafe boto3 session - returns a
+    ThreadLocalLazy wrapped around that callback.
+
+    This is the currently-preferred method of generating a threadsafe
+    client or resource at a module/singleton level.
+    """
+    return ThreadLocalLazy(lambda: resource_or_client_creator(_THREAD_SESSION()))
diff --git a/xoto3/sqs/utils.py b/xoto3/sqs/utils.py
index 60ee068..59013d6 100644
--- a/xoto3/sqs/utils.py
+++ b/xoto3/sqs/utils.py
@@ -1,7 +1,7 @@
 import json
 import typing as ty
 
-import boto3
+from xoto3.lazy_session import tll_from_session
 
 
 def yield_record_bodies_from_sqs_event(event: dict) -> ty.Iterable:
@@ -10,7 +10,13 @@ def yield_record_bodies_from_sqs_event(event: dict) -> ty.Iterable:
 
 
 def yield_and_ack_sqs_messages(
-    queue_url: str, limit: int = 0, accept=None, *, WaitTimeSeconds: int = 1, **kwargs
+    queue_url: str,
+    limit: int = 0,
+    accept=None,
+    *,
+    WaitTimeSeconds: int = 1,
+    sqs_client=None,
+    **kwargs,
 ) -> ty.Iterable[dict]:
     """Generates messages from an SQS queue.
 
@@ -21,7 +27,7 @@ def yield_and_ack_sqs_messages(
     :param queue_url: URL of the SQS queue to drain.
 
     """
-    sqs_client = boto3.client("sqs")
+    sqs_client = sqs_client or tll_from_session(lambda s: s.client("sqs"))()
     recvd = 0
 
     while recvd < limit or limit == 0:
diff --git a/xoto3/ssm/parameters.py b/xoto3/ssm/parameters.py
index 33182a3..be783d7 100644
--- a/xoto3/ssm/parameters.py
+++ b/xoto3/ssm/parameters.py
@@ -6,17 +6,17 @@
 Capital-letter arguments are SSM Parameter store arguments, whereas
 lowercase arguments are for our internal logic.
 """
-from typing import Union
-import re
+import json
 import os
 import os.path
-import json
+import re
 from datetime import datetime
-from typing_extensions import Literal
+from typing import Union
 
 from botocore.exceptions import ClientError
+from typing_extensions import Literal
 
-from xoto3.lazy_session import tlls
+from xoto3.lazy_session import tll_from_session
 
 ParameterType = Union[Literal["String"], Literal["SecureString"], Literal["StringList"]]
 
@@ -35,7 +35,7 @@
     SCRIPT_NAME = "SSM Params"
 
 
-_SSM_CLIENT = tlls("client", "ssm")
+_SSM_CLIENT = tll_from_session(lambda sess: sess.client("ssm"))
 
 
 def delete(Name: str, ssm=None) -> bool:


Test patch:
diff --git a/tests/xoto3/dynamodb/get_test.py b/tests/xoto3/dynamodb/get_test.py
index b522de6..3b5f2ba 100644
--- a/tests/xoto3/dynamodb/get_test.py
+++ b/tests/xoto3/dynamodb/get_test.py
@@ -5,6 +5,7 @@
 from xoto3.dynamodb.exceptions import ItemNotFoundException, get_item_exception_type
 from xoto3.dynamodb.get import (
     GetItem,
+    retry_notfound_consistent_read,
     strongly_consistent_get_item,
     strongly_consistent_get_item_if_exists,
 )
@@ -35,3 +36,53 @@ def test_get_exists(integration_test_id_table, integration_test_id_table_put):
     assert item == strongly_consistent_get_item_if_exists(
         integration_test_id_table, random_key, nicename="TestItem2"
     )
+
+
+def test_retry_get_with_consistent_read_if_it_exists():
+
+    the_item = dict(id="foo")
+
+    def _fake_get(**kw):
+        if not kw.get("ConsistentRead"):
+            raise ItemNotFoundException("ouch")
+        return the_item
+
+    test_get = retry_notfound_consistent_read(_fake_get)
+
+    assert test_get() == the_item
+
+
+def test_retry_get_with_consistent_read_if_it_does_not_exist():
+    calls = 0
+
+    def _fake_get(**kw):
+        nonlocal calls
+        calls += 1
+        if not kw.get("ConsistentRead"):
+            raise ItemNotFoundException("ouch")
+        raise ItemNotFoundException("definitely does not exist")
+
+    test_get = retry_notfound_consistent_read(_fake_get)
+
+    with pytest.raises(ItemNotFoundException):
+        test_get()
+
+    assert calls == 2
+
+
+def test_dont_retry_get_with_consistent_read_if_it_was_already_consistent():
+    calls = 0
+
+    def _fake_get(**kw):
+        nonlocal calls
+        calls += 1
+        if not kw.get("ConsistentRead"):
+            raise ItemNotFoundException("ouch")
+        raise ItemNotFoundException("definitely does not exist")
+
+    test_get = retry_notfound_consistent_read(_fake_get)
+
+    with pytest.raises(ItemNotFoundException):
+        test_get(ConsistentRead=True)
+
+    assert calls == 1
diff --git a/tests/xoto3/dynamodb/write_versioned/__init__.py b/tests/xoto3/dynamodb/write_versioned/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/xoto3/dynamodb/write_versioned/api2_test.py b/tests/xoto3/dynamodb/write_versioned/api2_test.py
new file mode 100644
index 0000000..b3b8cb8
--- /dev/null
+++ b/tests/xoto3/dynamodb/write_versioned/api2_test.py
@@ -0,0 +1,98 @@
+from random import randint
+
+import pytest
+
+from xoto3.dynamodb.exceptions import ItemNotFoundException
+from xoto3.dynamodb.write_versioned import (
+    ItemTable,
+    VersionedTransaction,
+    create_or_update,
+    update_existing,
+    update_if_exists,
+    versioned_transact_write_items,
+)
+
+from .conftest import mock_next_run
+
+
+def _randkey():
+    return dict(id="".join((str(randint(0, 9)) for i in range(30))))
+
+
+def test_single_table_helpers(integration_test_id_table):
+    itable = ItemTable(integration_test_id_table.name, item_name="Steve")
+
+    rand_key = _randkey()
+    # random because it's actually writing to the table
+
+    def transact(vt):
+        vt = itable.put(dict(rand_key, foo="bar"))(vt)
+        vt = itable.delete(rand_key)(vt)
+        return vt
+
+    out = versioned_transact_write_items(transact, {integration_test_id_table.name: [rand_key]})
+    assert None is itable.get(rand_key)(out)
+
+
+def _fake_table(
+    table_name: str, *key_attrs: str,
+):
+    table = ItemTable(table_name)
+    vt = VersionedTransaction(dict())
+    vt = table.define(*key_attrs)(vt)
+    return vt, table
+
+
+def _update(key, val):
+    def _(it):
+        it[key] = val
+        return it
+
+    return _
+
+
+def test_api2_update_if_exists():
+    vt, table = _fake_table(lambda: "steve", "id")  # type: ignore
+
+    key = dict(id="exists")
+    vt = table.presume(key, dict(key, foo="bar"))(vt)
+
+    vt = versioned_transact_write_items(
+        update_if_exists(table, _update("foo", "biz"), key), **mock_next_run(vt),
+    )
+
+    assert table.require(key)(vt)["foo"] == "biz"
+
+    bad_key = dict(id="notexists")
+    vt = versioned_transact_write_items(
+        update_if_exists(table, _update("foo", "zap"), bad_key),
+        dict(steve=[key]),
+        **mock_next_run(vt),
+    )
+    assert table.require(key)(vt)["foo"] == "biz"  # the same
+    assert None is table.get(bad_key)(vt)
+
+
+def test_api2_update_existing():
+    vt, table = _fake_table("steve", "id")
+    key = dict(id="yo")
+    vt = table.presume(key, dict(key, foo="blah"))(vt)
+
+    vt = versioned_transact_write_items(
+        update_existing(table, _update("foo", "zot"), key), **mock_next_run(vt)
+    )
+    assert table.require(key)(vt)["foo"] == "zot"
+
+    with pytest.raises(ItemNotFoundException):
+        versioned_transact_write_items(
+            update_existing(table, _update("foo", "oi"), dict(id="no"),), **mock_next_run(vt),
+        )
+
+
+def test_api2_create_or_update():
+    vt, table = _fake_table("steve", "id")
+    key = dict(id="yolo")
+    vt = versioned_transact_write_items(
+        create_or_update(table, lambda x: dict(key, foo=1), key), **mock_next_run(vt),
+    )
+    assert table.require(key)(vt)["foo"] == 1
diff --git a/tests/xoto3/dynamodb/write_versioned/conftest.py b/tests/xoto3/dynamodb/write_versioned/conftest.py
new file mode 100644
index 0000000..c91f245
--- /dev/null
+++ b/tests/xoto3/dynamodb/write_versioned/conftest.py
@@ -0,0 +1,48 @@
+from typing import Callable, cast
+
+from xoto3.dynamodb.write_versioned.keys import hashable_key
+from xoto3.dynamodb.write_versioned.types import (
+    BatchGetItem,
+    Item,
+    ItemKey,
+    ItemKeysByTableName,
+    ItemsByTableName,
+    Optional,
+    VersionedTransaction,
+    _TableData,
+)
+
+
+def _probe_table(table_data: _TableData, key: ItemKey) -> Optional[Item]:
+    hkey = hashable_key(key)
+    if hkey in table_data.effects:
+        return table_data.effects[hkey]
+    return table_data.items.get(hkey, None)
+
+
+def transaction_to_batch_getter(vt: VersionedTransaction) -> BatchGetItem:
+    """Take a generated transaction and turn it back into an input for a faked transaction run."""
+
+    def batch_get_item(item_keys: ItemKeysByTableName, **_kwargs) -> ItemsByTableName:
+        return {
+            table_name: list(
+                filter(
+                    None,
+                    [_probe_table(vt.tables[table_name], key) for key in item_keys[table_name]],
+                )
+            )
+            if table_name in vt.tables
+            else list()
+            for table_name in item_keys
+        }
+
+    return cast(BatchGetItem, batch_get_item)
+
+
+def mock_next_run(
+    vt: VersionedTransaction,
+    transact_write_items: Callable[[VersionedTransaction], None] = lambda **_kw: None,
+):
+    return dict(
+        batch_get_item=transaction_to_batch_getter(vt), transact_write_items=transact_write_items
+    )
diff --git a/tests/xoto3/dynamodb/write_versioned/example_unit_test.py b/tests/xoto3/dynamodb/write_versioned/example_unit_test.py
new file mode 100644
index 0000000..3b68510
--- /dev/null
+++ b/tests/xoto3/dynamodb/write_versioned/example_unit_test.py
@@ -0,0 +1,75 @@
+"""Imagine you had the following business logic and you wanted to test
+it. Here's how you might go about that, using the
+xoto3.dynamodb.write_versioned API itself to set everything up and
+prove that it worked.
+
+The example here is the same as the code we have in the README.
+"""
+import pytest
+
+import xoto3.dynamodb.write_versioned as wv
+
+# threadsafe because they are pure values.
+task_table = wv.ItemTable("TaskTable", item_name="Task")
+user_table = wv.ItemTable("UserTable", item_name="User")
+
+# imagine these were passed in via an API
+task_key = dict(id="task1")
+new_task = dict(task_key, name="plant tree", num_subtasks=3, user_id="steve")
+
+# this transaction builder is a pure function that constructs
+# a value representing DynamoDB effects to be transactionally/atomically applied.
+def create_and_link_task_unless_exists(t: wv.VersionedTransaction) -> wv.VersionedTransaction:
+    t = task_table.presume(task_key, None)(t)
+    # ^ we presume that the task does not exist, i.e. has the value None
+    # This is a no-op if a value has already been fetched.
+    # This is also purely an optimization to avoid an initial read;
+    # the whole transaction would result in the same data in the table without it.
+
+    existing_task = task_table.get(task_key)(t)
+    if existing_task:
+        return t
+        # ^ perform no action as long as the item already exists
+
+    t = task_table.put(new_task)(t)
+    # ^ put this item into this table as long as the item does not exist
+    user = user_table.require(dict(id=new_task["user_id"]))(t)  # type: ignore
+    # ^ fetch the user and raise an exception if it does not exist
+    user["task_ids"].append(task_key["id"])
+    t = user_table.put(user)(t)
+    # ^ make sure that the user knows about its task
+    return t
+    # ^ return the built transaction to be executed.
+
+
+def test_task_and_user_are_not_written_if_task_already_exists():
+    t = wv.VersionedTransaction(dict())
+    existing_task = dict(task_key, name="water flowers", is_done=True, user_id="felicity")
+    t = task_table.presume(task_key, existing_task)(t)
+    # presume lets us set up a fixture in the fake table, represented by the VersionedTransaction
+
+    t = create_and_link_task_unless_exists(t)
+
+    assert existing_task == task_table.require(task_key)(t)
+    # the task remains unchanged
+
+    with pytest.raises(wv.ItemUndefinedException):
+        user_table.get(dict(id="felicity"))(t)
+        # the user was not even fetched, much less written
+
+
+def test_task_and_user_are_both_written_if_task_does_not_exist():
+    t = wv.VersionedTransaction(dict())
+    t = task_table.presume(task_key, None)(t)
+    # declare that the task does not exist - essentially this is a 'fixture', but without I/O
+    user_key = dict(id="steve")
+    user = dict(user_key, name="Actually Steve", task_ids=list())
+    t = user_table.presume(user_key, user)(t)
+
+    t = create_and_link_task_unless_exists(t)
+
+    resulting_task = task_table.require(task_key)(t)
+    resulting_user = user_table.require(user_key)(t)
+    assert resulting_task == new_task
+    assert resulting_user["name"] == "Actually Steve"
+    assert resulting_user["task_ids"] == ["task1"]
diff --git a/tests/xoto3/dynamodb/write_versioned/transact/build_test.py b/tests/xoto3/dynamodb/write_versioned/transact/build_test.py
index 6bba1bf..59d6288 100644
--- a/tests/xoto3/dynamodb/write_versioned/transact/build_test.py
+++ b/tests/xoto3/dynamodb/write_versioned/transact/build_test.py
@@ -2,7 +2,7 @@
 
 from xoto3.dynamodb.exceptions import ItemNotFoundException
 from xoto3.dynamodb.write_versioned import delete, get, put, require
-from xoto3.dynamodb.write_versioned.errors import ItemNotYetFetchedError, TableSchemaUnknownError
+from xoto3.dynamodb.write_versioned.errors import ItemUndefinedException, TableSchemaUnknownError
 from xoto3.dynamodb.write_versioned.keys import hashable_key, key_from_item
 from xoto3.dynamodb.write_versioned.prepare import items_and_keys_to_clean_table_data
 from xoto3.dynamodb.write_versioned.types import VersionedTransaction as VT
@@ -34,10 +34,10 @@ def test_get_and_require():
     with pytest.raises(ItemNotFoundException):
         require(vt, "table1", dict(id="no"))
 
-    with pytest.raises(ItemNotYetFetchedError):
+    with pytest.raises(ItemUndefinedException):
         get(vt, "table2", dict(id="1"),) == dict(id="1", val="a")
 
-    with pytest.raises(ItemNotYetFetchedError):
+    with pytest.raises(ItemUndefinedException):
         get(vt, "table1", dict(id="3"))
 
 
diff --git a/tests/xoto3/dynamodb/write_versioned/transact/lazy_batch_gets_test.py b/tests/xoto3/dynamodb/write_versioned/transact/lazy_batch_gets_test.py
new file mode 100644
index 0000000..fd331f6
--- /dev/null
+++ b/tests/xoto3/dynamodb/write_versioned/transact/lazy_batch_gets_test.py
@@ -0,0 +1,48 @@
+from xoto3.dynamodb.write_versioned import (
+    ItemTable,
+    VersionedTransaction,
+    versioned_transact_write_items,
+)
+
+
+def test_batch_get_lazy_load():
+    """Tests that multiple serial `get`s perform only a single actual call
+    to batch_get"""
+    t = VersionedTransaction(dict())
+    table_a = ItemTable("a")
+    table_b = ItemTable("b")
+
+    a1_k = dict(id="a1")
+    a2_k = dict(id="a2")
+    b1_k = dict(id="b1")
+
+    a3_k = dict(id="a3")
+
+    def triple_get(t: VersionedTransaction) -> VersionedTransaction:
+        a1 = table_a.get(a1_k)(t)
+        b1 = table_b.get(b1_k)(t)
+        a2 = table_a.get(a2_k)(t)
+        # all three of the above gets will be performed
+        # together as a single call to batch_get.
+        return table_a.put(dict(a3_k, items=[a1, b1, a2]))(t)
+
+    calls = 0
+    a1 = dict(a1_k, i=6)
+    a2 = dict(a2_k, i=8)
+    b1 = dict(b1_k, j=4)
+
+    def batch_get(item_keys_by_table_name):
+        if not item_keys_by_table_name:
+            return dict()
+        nonlocal calls
+        calls += 1
+        return dict(a=[a1, a2], b=[b1])
+
+    t = versioned_transact_write_items(
+        triple_get,
+        batch_get_item=batch_get,  # type: ignore
+        transact_write_items=lambda **_kw: None,
+    )
+
+    assert calls == 1
+    assert table_a.require(a3_k)(t)["items"] == [a1, b1, a2]
diff --git a/tests/xoto3/dynamodb/write_versioned/transact/retry_test.py b/tests/xoto3/dynamodb/write_versioned/transact/retry_test.py
index f3c1758..720f2be 100644
--- a/tests/xoto3/dynamodb/write_versioned/transact/retry_test.py
+++ b/tests/xoto3/dynamodb/write_versioned/transact/retry_test.py
@@ -1,4 +1,5 @@
 import time
+from datetime import timedelta
 
 from xoto3.dynamodb.write_versioned.retry import timed_retry
 
@@ -7,7 +8,7 @@ def test_timed_retry():
     attempted = 0
 
     started_at = time.monotonic()
-    for _ in timed_retry(0.5, 7):
+    for _ in timed_retry(timedelta(seconds=0.5), 7):
         attempted += 1
     assert attempted == 7
     assert 0.3 < time.monotonic() - started_at < 0.5
diff --git a/tests/xoto3/dynamodb/write_versioned/transact/specify_test.py b/tests/xoto3/dynamodb/write_versioned/transact/specify_test.py
new file mode 100644
index 0000000..4f9fd51
--- /dev/null
+++ b/tests/xoto3/dynamodb/write_versioned/transact/specify_test.py
@@ -0,0 +1,39 @@
+from xoto3.dynamodb.write_versioned import VersionedTransaction, get, put, require
+from xoto3.dynamodb.write_versioned.keys import hashable_key
+from xoto3.dynamodb.write_versioned.modify import delete
+from xoto3.dynamodb.write_versioned.specify import define_table, presume
+from xoto3.dynamodb.write_versioned.types import _TableData
+
+
+def test_presume():
+    tx = VersionedTransaction(dict())
+    tx = presume(tx, "steve", dict(group="g", id="123"), None)
+    assert None is get(tx, "steve", dict(group="g", id="123"))
+
+
+def test_presume_already_exists():
+    key = dict(id=123)
+    tx = VersionedTransaction(
+        dict(
+            steve=_TableData(
+                items={hashable_key(key): dict(key, foo=2)}, effects=dict(), key_attributes=("id",)
+            )
+        )
+    )
+    tx = presume(tx, "steve", key, None)
+    assert require(tx, "steve", dict(id=123))["foo"] == 2
+
+
+def test_define_table():
+    tx = VersionedTransaction(dict())
+    tx = define_table(tx, "BobTable", "group", "id")
+    item_key = dict(group=1, id=2)
+    item = dict(item_key, foo=7)
+    tx = put(tx, "BobTable", item)
+    assert require(tx, "BobTable", item_key) == item
+
+
+def test_define_table_doesnt_redefine():
+    tx = delete(VersionedTransaction(dict()), "table1", dict(id="foo"))
+    same_tx = define_table(tx, "table1", "id")
+    assert same_tx is tx