Merge d46d4ad into 964da77

.gitignore

@@ -5,3 +5,4 @@ kinto_http.egg-info/
 .venv/
 .tox
 .pytest_cache/
+.vscode/

Makefile

@@ -1,4 +1,4 @@
-VIRTUALENV = virtualenv --python=python3.7
+VIRTUALENV = virtualenv --python=python3
 VENV := $(shell echo $${VIRTUAL_ENV-.venv})
 PYTHON = $(VENV)/bin/python
 DEV_STAMP = $(VENV)/.dev_env_installed.stamp
@@ -34,13 +34,13 @@ runkinto: install-dev
 	$(VENV)/bin/kinto start --ini kinto_http/tests/config/kinto.ini
 
 tests-once: install-dev need-kinto-running
-	$(VENV)/bin/py.test kinto_http/tests/functional.py kinto_http/tests --cov-report term-missing --cov-fail-under 100 --cov kinto_http
+	$(VENV)/bin/pytest --cov-report term-missing --cov-fail-under 100 --cov kinto_http
 
 functional: install-dev need-kinto-running
-	$(VENV)/bin/py.test kinto_http/tests/functional.py
+	$(VENV)/bin/pytest -k "test_functional"
 
 tests: install-dev need-kinto-running
-	$(VENV)/bin/py.test -f kinto_http/tests/ kinto_http/tests/functional.py
+	$(VENV)/bin/pytest
 
 format: install-dev
 	$(VENV)/bin/isort --profile=black --lines-after-imports=2 kinto_http

dev-requirements.txt

@@ -1,10 +1,12 @@
 black
 flake8
 isort
+kinto
 pytest
+pytest-asyncio
 pytest-cache
 pytest-cov
+pytest-mock
 pytest-xdist
-kinto
 therapist
 unidecode

kinto_http/aio.py

@@ -0,0 +1,275 @@
+import asyncio
+import logging
+import random
+from typing import Dict, List
+
+import backoff
+import requests
+
+from kinto_http import Client, utils
+
+
+logger = logging.getLogger(__name__)
+
+retry_timeout = backoff.on_exception(
+    backoff.expo,
+    (requests.exceptions.Timeout, requests.exceptions.ConnectionError),
+    max_tries=2,
+)
+
+
+class AsyncClient(Client):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._client = Client(*args, **kwargs)
+
+    def clone(self, **kwargs):
+        if "server_url" in kwargs or "auth" in kwargs:
+            kwargs.setdefault("server_url", self.session.server_url)
+            kwargs.setdefault("auth", self.session.auth)
+        else:
+            kwargs.setdefault("session", self.session)
+        kwargs.setdefault("bucket", self._bucket_name)
+        kwargs.setdefault("collection", self._collection_name)
+        kwargs.setdefault("retry", self.session.nb_retry)
+        kwargs.setdefault("retry_after", self.session.retry_after)
+        return AsyncClient(**kwargs)
+
+    @retry_timeout
+    async def server_info(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.server_info(*args, **kwargs))
+
+    @retry_timeout
+    async def get_bucket(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.get_bucket(*args, **kwargs))
+
+    @retry_timeout
+    async def get_buckets(self, **kwargs) -> List[Dict]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.get_buckets(**kwargs))
+
+    @retry_timeout
+    async def create_bucket(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.create_bucket(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def update_bucket(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.update_bucket(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def patch_bucket(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.patch_bucket(*args, **kwargs))
+
+    @retry_timeout
+    async def delete_bucket(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.delete_bucket(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def delete_buckets(self, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.delete_buckets(**kwargs))
+
+    @retry_timeout
+    async def get_group(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.get_group(*args, **kwargs))
+
+    @retry_timeout
+    async def get_groups(self, *args, **kwargs) -> List[Dict]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.get_groups(*args, **kwargs))
+
+    @retry_timeout
+    async def create_group(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.create_group(*args, **kwargs))
+
+    @retry_timeout
+    async def update_group(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.update_group(*args, **kwargs))
+
+    @retry_timeout
+    async def patch_group(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.patch_group(*args, **kwargs))
+
+    @retry_timeout
+    async def delete_group(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.delete_group(*args, **kwargs))
+
+    @retry_timeout
+    async def delete_groups(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.delete_groups(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def get_collection(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.get_collection(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def get_collections(self, *args, **kwargs) -> List[Dict]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.get_collections(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def create_collection(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.create_collection(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def update_collection(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.update_collection(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def patch_collection(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.patch_collection(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def delete_collection(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.delete_collection(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def delete_collections(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.delete_collections(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def get_record(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.get_record(*args, **kwargs))
+
+    @retry_timeout
+    async def get_records(self, *args, **kwargs) -> List[Dict]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.get_records(*args, **kwargs))
+
+    @retry_timeout
+    async def get_paginated_records(self, *args, **kwargs) -> List[Dict]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.get_paginated_records(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def get_records_timestamp(self, *args, **kwargs) -> str:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.get_records_timestamp(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def create_record(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.create_record(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def update_record(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.update_record(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def patch_record(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.patch_record(*args, **kwargs))
+
+    @retry_timeout
+    async def delete_record(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.delete_record(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def delete_records(self, *args, **kwargs) -> Dict:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.delete_records(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def get_history(self, *args, **kwargs) -> List[Dict]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.get_history(*args, **kwargs))
+
+    @retry_timeout
+    async def purge_history(self, *args, **kwargs) -> List[Dict]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            None, lambda: self._client.purge_history(*args, **kwargs)
+        )
+
+    @retry_timeout
+    async def get_endpoint(self, *args, **kwargs) -> str:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: self._client.get_endpoint(*args, **kwargs))
+
+    @retry_timeout
+    async def get_monitor_changes(self, bust_cache=False, **kwargs) -> List[Dict]:
+        if bust_cache:
+            if "_expected" in kwargs:
+                raise ValueError("Pick one of `bust_cache` and `_expected` parameters")
+            random_cache_bust = random.randint(999999000000, 999999999999)
+            kwargs["_expected"] = random_cache_bust
+        return await self.get_records(bucket="monitor", collection="changes", **kwargs)
+
+    # TODO: get proper tests written for this
+    # @retry_timeout
+    # async def get_changeset(self, bucket, collection, **kwargs) -> List[Dict]:
+    #     endpoint = f"/buckets/{bucket}/collections/{collection}/changeset"
+    #     kwargs.setdefault("_expected", random.randint(999999000000, 999999999999))
+    #     loop = asyncio.get_event_loop()
+    #     body, _ = await loop.run_in_executor(
+    #         None, lambda: self._client.session.request("get", endpoint, params=kwargs)
+    #     )
+    #     return body
+
+    def __repr__(self):
+        if self._collection_name:
+            endpoint = self._client.get_endpoint(
+                "collection", bucket=self._bucket_name, collection=self._collection_name
+            )
+        elif self._bucket_name:
+            endpoint = self._client.get_endpoint("bucket", bucket=self._bucket_name)
+        else:
+            endpoint = self._client.get_endpoint("root")
+
+        absolute_endpoint = utils.urljoin(self.session.server_url, endpoint)
+        return f"<KintoClient {absolute_endpoint}>"

kinto_http/tests/config/kinto.ini

@@ -6,6 +6,7 @@ kinto.paginate_by = 5
 
 kinto.includes = kinto.plugins.flush
                  kinto.plugins.accounts
+                 kinto_changes
 
 multiauth.policies = account
 multiauth.policy.account.use = kinto.plugins.accounts.AccountsPolicy

kinto_http/tests/conftest.py

@@ -0,0 +1,44 @@
+from typing import Tuple
+from unittest.mock import MagicMock
+from urllib.parse import urljoin
+
+import pytest
+import requests
+from pytest_mock.plugin import MockerFixture
+
+from kinto_http.aio import AsyncClient as Client
+
+from .support import mock_response
+from .test_functional_async import DEFAULT_AUTH, SERVER_URL, create_user
+
+
+@pytest.fixture
+def async_client_setup(mocker: MockerFixture) -> Tuple[Client, MagicMock]:
+    session = mocker.MagicMock()
+    mock_response(session)
+    client = Client(session=session, bucket="mybucket")
+    return client, session
+
+
+@pytest.fixture
+def record_setup(mocker: MockerFixture) -> Tuple[Client, MagicMock]:
+    session = mocker.MagicMock()
+    session.request.return_value = (mocker.sentinel.response, mocker.sentinel.count)
+    client = Client(session=session, bucket="mybucket", collection="mycollection")
+    return client, session
+
+
+@pytest.fixture
+def functional_setup():
+    # Setup
+    # Create user and return client
+    client = Client(server_url=SERVER_URL, auth=DEFAULT_AUTH)
+    create_user(SERVER_URL, DEFAULT_AUTH)
+
+    yield client
+
+    # Teardown
+    # Delete all the created objects
+    flush_url = urljoin(SERVER_URL, "/__flush__")
+    resp = requests.post(flush_url)
+    resp.raise_for_status()