Issue with latest moto server with aiobotocore

[Manisha28696]

I'm using the latest moto server v5.0.2 which is giving KeyError: 'Table' during the describe dynamodb table whereas, create_table succeeded with 200 response with missing TableDescription attribute in it. can someone help me with what I'm doing wrong here?

my_req.txt

Here is my Moto Service which I created

moto_server.py

import asyncio
import functools
import logging
import socket
import threading
import time
import os

# Third Party
import aiohttp
import moto.server
import werkzeug.serving


host = '127.0.0.1'

_PYCHARM_HOSTED = os.environ.get('PYCHARM_HOSTED') == '1'
_CONNECT_TIMEOUT = 90 if _PYCHARM_HOSTED else 10


def get_free_tcp_port(release_socket: bool = False):
    sckt = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sckt.bind((host, 0))
    addr, port = sckt.getsockname()
    if release_socket:
        sckt.close()
        return port

    return sckt, port


class MotoService:
    """ Will Create MotoService.
    Service is ref-counted so there will only be one per process. Real Service will
    be returned by `__aenter__`."""

    def __init__(self, port: int = None):
        if port:
            self._socket = None
            self._port = port
        else:
            self._socket, self._port = get_free_tcp_port()

        self._thread = None
        self._logger = logging.getLogger('MotoService')
        self._refcount = None
        self._ip_address = host
        self._server = None

    @property
    def endpoint_url(self):
        return 'http://{}:{}'.format(self._ip_address, self._port)

    def __call__(self, func):
        async def wrapper(*args, **kwargs):
            await self._start()
            try:
                result = await func(*args, **kwargs)
            finally:
                await self._stop()
            return result

        functools.update_wrapper(wrapper, func)
        wrapper.__wrapped__ = func
        return wrapper

    async def __aenter__(self):
        if self._refcount is None:
            self._refcount = 1
            await self._start()
            return self
        else:
            self._refcount += 1
            return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        self._refcount -= 1

        if self._socket:
            self._socket.close()
            self._socket = None

        if self._refcount == 0:
            await self._stop()

    def _server_entry(self):
        self._main_app = moto.server.DomainDispatcherApplication(
            moto.server.create_backend_app)
        self._main_app.debug = True

        if self._socket:
            self._socket.close()  # release right before we use it
            self._socket = None

        self._server = werkzeug.serving.make_server(
            self._ip_address, self._port, self._main_app, True)
        self._server.serve_forever()

    async def _start(self):
        self._thread = threading.Thread(target=self._server_entry, daemon=True)
        self._thread.start()

        async with aiohttp.ClientSession() as session:
            start = time.time()

            while time.time() - start < 10:
                if not self._thread.is_alive():
                    break

                try:
                    # we need to bypass the proxies due to monkeypatches
                    async with session.get(self.endpoint_url + '/static',
                                           timeout=_CONNECT_TIMEOUT):
                        pass
                    break
                except (asyncio.TimeoutError, aiohttp.ClientConnectionError):
                    await asyncio.sleep(0.5)
            else:
                await self._stop()  # pytest.fail doesn't call stop_process
                raise Exception("Can not start service: {}".format(
                    self._service_name))

    async def _stop(self):
        if self._server:
            self._server.shutdown()

        self._thread.join()

test_moto_server.py

import aiobotocore
import pytest
from aiobotocore.config import AioConfig
from .moto_server import MotoService


@pytest.fixture(scope='session')
async def aws_server_moto():
    async with MotoService() as svc:
        yield svc.endpoint_url


@pytest.fixture(scope='session')
def session_moto():
    session = aiobotocore.session.AioSession()
    return session


@pytest.fixture(scope='session')
def region_moto():
    return 'us-east-1'


@pytest.fixture(scope='session')
def signature_version_moto():
    return 's3'

@pytest.fixture(scope="session")
def moto_config_moto(aws_server_moto):
    kw = dict(endpoint_url=aws_server_moto,
              aws_secret_access_key="xxx",
              aws_access_key_id="xxx")
    return kw

@pytest.fixture(scope='session')
def config_moto(region_moto, signature_version_moto):
    connect_timeout = read_timout = 5
    return AioConfig(region_name=region_moto, signature_version=signature_version_moto,
                     read_timeout=read_timout, connect_timeout=connect_timeout)

@pytest.fixture(scope='session')
async def dynamo_client(session_moto, region_moto, config_moto, moto_config_moto):
    async with session_moto.create_client('dynamodb', region_name=region_moto,
                                     config=config_moto, **moto_config_moto) as client:
        yield client


async def test_mock_infra(dynamo_client):
    async def _is_table_ready(table_name):
        response = await dynamo_client.describe_table(
            TableName=table_name
        )
        return response['Table']['TableStatus'] == 'ACTIVE'

    # Creating dynamodb task table
    key_schema = [{'AttributeName': 'request_id', 'KeyType': 'HASH'}]
    attr_definition = [{'AttributeName': 'request_id', 'AttributeType': 'S'}]
    table_name = 'test_table'
    table_kwargs = {
        'TableName': table_name,
        'AttributeDefinitions': attr_definition,
        'KeySchema': key_schema,
        'BillingMode': 'PAY_PER_REQUEST',
    }

    response = await dynamo_client.create_table(**table_kwargs)
    while not (await _is_table_ready(table_name)):
        pass

    assert response['ResponseMetadata']['HTTPStatusCode'] == 200

[bblommers]

Hi @Manisha28696, on a general note: you would have to install moto[server], not just the moto-package, to ensure it pulls in all the required dependencies.

Having said that - I cannot try this out, because I'm unable to install the requirements that you've provided:

ERROR: Could not find a version that satisfies the requirement pywin32>=223 (from pypiwin32) (from versions: none)
ERROR: No matching distribution found for pywin32>=223

Even with a simplified requirements-file:

aioboto3==12.3.0
aiobotocore==2.11.2
aiohttp==3.9.3
aioitertools==0.11.0
aiosignal==1.3.1
moto[server]==5.0.2

I'm running into an issue:

AttributeError: module 'aiobotocore' has no attribute 'session'

[Manisha28696]

I'm able to do aiobotocore.session.AioSession() for given package versions. Also after debugging found out that changing signature_version from s3 to s3v4 in AioConfig has worked for me.

[bblommers]

Also after debugging found out that changing signature_version from s3 to s3v4 in AioConfig has worked for me.

Do you mean that solved the problem? Great, if that's the case, I'll close this ticket. Thanks for letting us know the solution!