complete base conn interface

Author: garrettheel

pynamodb/async_util.py

@@ -0,0 +1,19 @@
+import functools
+
+
+def run_secretly_sync_async_fn(async_fn, *args, **kwargs):
+    # From https://github.com/python-trio/hip/issues/1#issuecomment-322028457
+    coro = async_fn(*args, **kwargs)
+    try:
+        coro.send(None)
+    except StopIteration as exc:
+        return exc.value
+    else:
+        raise RuntimeError("you lied, this async function is not secretly synchronous")
+
+
+def wrap_secretly_sync_async_fn(async_fn):
+    @functools.wraps(async_fn)
+    def wrap(*args, **kwargs):
+        return run_secretly_sync_async_fn(async_fn, *args, **kwargs)
+    return wrap

pynamodb/connection/base.py

@@ -2,7 +2,6 @@
 Lowest level connection
 """
 import asyncio
-import functools
 import json
 import logging
 import random
@@ -28,6 +27,7 @@
     from aiobotocore.session import get_session as get_async_session
 
 
+from pynamodb.async_util import wrap_secretly_sync_async_fn
 from pynamodb.constants import (
     RETURN_CONSUMED_CAPACITY_VALUES, RETURN_ITEM_COLL_METRICS_VALUES,
     RETURN_ITEM_COLL_METRICS, RETURN_CONSUMED_CAPACITY, RETURN_VALUES_VALUES,
@@ -71,17 +71,6 @@
 log.addHandler(logging.NullHandler())
 
 
-def run_secretly_sync_async_fn(async_fn, *args, **kwargs):
-    # From https://github.com/python-trio/hip/issues/1#issuecomment-322028457
-    coro = async_fn(*args, **kwargs)
-    try:
-        coro.send(None)
-    except StopIteration as exc:
-        return exc.value
-    else:
-        raise RuntimeError("you lied, this async function is not secretly synchronous")
-
-
 class MetaTable(object):
     """
     A pythonic wrapper around table metadata
@@ -254,20 +243,15 @@ def get_exclusive_start_key_map(self, exclusive_start_key):
             }
 
 
-def sync_wrapper(async_fn):
-    @functools.wraps(async_fn)
-    def wrap(*args, **kwargs):
-        return run_secretly_sync_async_fn(async_fn, *args, **kwargs)
-    return wrap
-
-
 class ConnectionMeta(type):
     def __init__(self, name, bases, attrs):
         super().__init__(name, bases, attrs)
 
         for attr_name, attr_value in attrs.items():
             if attr_name.endswith('_async') and asyncio.iscoroutinefunction(attr_value):
-                setattr(self, attr_name.rstrip("_async"), sync_wrapper(attr_value))
+                wrapped_fn = wrap_secretly_sync_async_fn(attr_value)
+                wrapped_fn.__name__ = wrapped_fn.__name__.rstrip('_async')
+                setattr(self, wrapped_fn.__name__, wrapped_fn)
 
 
 class Connection(metaclass=ConnectionMeta):
@@ -370,7 +354,7 @@ async def dispatch(self, operation_name, operation_kwargs):
 
         self.send_pre_boto_callback(operation_name, req_uuid, table_name)
 
-        data = self._make_api_call(operation_name, operation_kwargs)
+        data = await self._make_api_call(operation_name, operation_kwargs)
         if asyncio.iscoroutine(data):
             data = await data
 
@@ -395,7 +379,7 @@ def send_pre_boto_callback(self, operation_name, req_uuid, table_name):
         except Exception as e:
             log.exception("pre_boto callback threw an exception.")
 
-    def _make_api_call(self, operation_name, operation_kwargs):
+    async def _make_api_call(self, operation_name, operation_kwargs):
         """
         This private method is here for two reasons:
         1. It's faster to avoid using botocore's response parsing
@@ -422,7 +406,7 @@ def _make_api_call(self, operation_name, operation_kwargs):
                     prepared_request.reset_stream()
 
                 # Create a new request for each retry (including a new signature).
-                prepared_request = run_secretly_sync_async_fn(self._create_prepared_request, self.client, request_dict, operation_model)
+                prepared_request = await self._create_prepared_request(self.client, request_dict, operation_model)
 
                 # Implement the before-send event from botocore
                 event_name = 'before-send.dynamodb.{}'.format(operation_model.name)
@@ -694,7 +678,7 @@ async def create_table_async(
             raise TableError("Failed to create table: {}".format(e), e)
         return data
 
-    def update_time_to_live(self, table_name: str, ttl_attribute_name: str) -> Dict:
+    async def update_time_to_live_async(self, table_name: str, ttl_attribute_name: str) -> Dict:
         """
         Performs the UpdateTimeToLive operation
         """
@@ -706,7 +690,7 @@ def update_time_to_live(self, table_name: str, ttl_attribute_name: str) -> Dict:
             }
         }
         try:
-            return self.dispatch(UPDATE_TIME_TO_LIVE, operation_kwargs)
+            return await self.dispatch(UPDATE_TIME_TO_LIVE, operation_kwargs)
         except BOTOCORE_EXCEPTIONS as e:
             raise TableError("Failed to update TTL on table: {}".format(e), e)
 

pynamodb/connection/table.py

@@ -3,15 +3,26 @@
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 """
 
+import asyncio
 from typing import Any, Dict, Mapping, Optional, Sequence
 
+from pynamodb.async_util import wrap_secretly_sync_async_fn
 from pynamodb.connection.base import Connection, MetaTable
 from pynamodb.constants import DEFAULT_BILLING_MODE, KEY
 from pynamodb.expressions.condition import Condition
 from pynamodb.expressions.update import Action
 
 
-class TableConnection:
+class TableMeta(type):
+    def __init__(self, name, bases, attrs):
+        super().__init__(name, bases, attrs)
+
+        for attr_name, attr_value in attrs.items():
+            if attr_name.endswith('_async') and asyncio.iscoroutinefunction(attr_value):
+                setattr(self, attr_name.rstrip("_async"), wrap_secretly_sync_async_fn(attr_value))
+
+
+class TableConnection(metaclass=TableMeta):
     """
     A higher level abstraction over botocore
     """
@@ -32,6 +43,8 @@ def __init__(
         aws_session_token: Optional[str] = None,
     ) -> None:
         self.table_name = table_name
+
+        # TODO: optional async
         self.connection = Connection(region=region,
                                      host=host,
                                      connect_timeout_seconds=connect_timeout_seconds,
@@ -46,7 +59,7 @@ def __init__(
                                                     aws_secret_access_key,
                                                     aws_session_token)
 
-    def get_meta_table(self, refresh: bool = False) -> MetaTable:
+    def get_meta_table_async(self, refresh: bool = False) -> MetaTable:
         """
         Returns a MetaTable
         """
@@ -259,11 +272,11 @@ def query(
             scan_index_forward=scan_index_forward,
             select=select)
 
-    def describe_table(self) -> Dict:
+    async def describe_table_async(self) -> Dict:
         """
         Performs the DescribeTable operation and returns the result
         """
-        return self.connection.describe_table(self.table_name)
+        return await self.connection.describe_table_async(self.table_name)
 
     def delete_table(self) -> Dict:
         """