Merge pull request #6 from aioworkers/connect

Connect

.travis.yml

@@ -9,11 +9,12 @@ python:
 - 3.7
 install:
 - pip install -U pip setuptools
-- pip install pytest-runner
+- pip install pipenv
+- pipenv install -d --skip-lock --system
 before_script:
 - psql -c 'create database test;' -U postgres
 script:
-- python setup.py test
+- pytest
 before_deploy:
 - echo "__version__ = '$(git describe --tags)'" > aioworkers_pg/version.py
 deploy:

Pipfile

@@ -4,16 +4,16 @@ verify_ssl = true
 name = "pypi"
 
 [packages]
-aioworkers = "*"
+aioworkers = ">=0.15"
 asyncpgsa = "*"
 
 [dev-packages]
-
 pytest = "*"
 pytest-aioworkers = "*"
 pytest-flake8 = "*"
-pytest-runner = "*"
 flake8-isort = "*"
+pyyaml = "*"
+mypy = "*"
 
 [requires]
 python_version = "3.7"

aioworkers_pg/base.py

@@ -1,37 +1,41 @@
-
-import logging
-
 import asyncpg
-from aioworkers.core.base import AbstractEntity
-
-logger = logging.getLogger('aioworkers_pg')
-
-
-class Connector(AbstractEntity):
-    # Method from pool which will be bind to connector
-    __bind_methods = (
-        'execute',
-        'executemany',
-        'fetch',
-        'fetchval',
-        'fetchrow',
-        'acquire',
-        'release',
-        'close',
-        'release',
-        'terminate',
-    )
-
-    def __init__(self, config=None, *, context=None, loop=None):
-        super().__init__(config, context=context, loop=loop)
-        self._pool = None
-        self.context.on_stop.append(self.stop)
+from aioworkers.core.base import AbstractConnector
 
-    async def stop(self):
-        await self._pool.close()
 
-    async def _create_pool(self):
-        return await asyncpg.create_pool(self.config.dsn, init=self._init_connection)
+class Connector(AbstractConnector):
+    def __init__(self, *args, **kwargs):
+        self._pool = None
+        super().__init__(*args, **kwargs)
+
+    def set_config(self, config):
+        cfg = config.new_parent(logger='aioworkers_pg')
+        super().set_config(cfg)
+
+    @property
+    def pool(self) -> asyncpg.pool.Pool:
+        assert self._pool
+        return self._pool
+
+    def __getattr__(self, attr):
+        # Proxy all unresolved attributes to the wrapped Pool object.
+        return getattr(self._pool, attr)
+
+    async def connect(self):
+        if self._pool is None:
+            self._pool = await self.pool_factory(self.config)
+
+    async def pool_factory(self, config):
+        pool = await asyncpg.create_pool(
+            config.dsn, init=self._init_connection,
+        )
+        self.logger.debug('Create pool with address %s', config.dsn)
+        return pool
+
+    async def disconnect(self):
+        if self._pool is not None:
+            self.logger.debug('Close pool')
+            await self._pool.close()
+            self._pool = None
 
     async def _init_connection(self, connection):
         import json
@@ -45,11 +49,3 @@ async def _init_connection(self, connection):
                 decoder=json.loads,
                 schema='pg_catalog',
             )
-
-    async def init(self):
-        await super().init()
-        self._pool = await self._create_pool()
-        for method_name in self.__bind_methods:
-            f = getattr(self._pool, method_name)
-            if f:
-                setattr(self, method_name, f)

aioworkers_pg/sa.py

@@ -1,11 +1,15 @@
+# true
 from .base import Connector as BaseConnector
 
 
 class Connector(BaseConnector):
-
-    async def _create_pool(self):
+    async def pool_factory(self, config):
         import asyncpgsa
-        return await asyncpgsa.create_pool(self.config.dsn, init=self._init_connection)
+        pool = await asyncpgsa.create_pool(
+            config.dsn, init=self._init_connection,
+        )
+        self.logger.debug('Create pool with address %s', config.dsn)
+        return pool
 
     async def _init_connection(self, connection):
         import json

aioworkers_pg/sql.py

@@ -2,6 +2,7 @@
 from collections import ChainMap
 from typing import Any, Match, Sequence
 
+
 NAME = re.compile(r'[^:]:([\d\w_]+)')
 
 
@@ -39,7 +40,7 @@ def with_data(self, *args, **kwargs):
     @staticmethod
     def _get_params(sql: str) -> Sequence[Any]:
         result = [sql]
-        pos = {}
+        pos = {}  # type: dict
 
         def repl(m: Match) -> str:
             name = m.group(1)

aioworkers_pg/storage.py

@@ -1,11 +1,18 @@
 from aioworkers.storage.base import AbstractStorageReadOnly
 
+# true
 from .base import Connector
 from .formatter import PGFormattedEntity
 from .sql import SQL, Table
 
 
 class RoStorage(PGFormattedEntity, Connector, AbstractStorageReadOnly):
+    def __init__(self, *args, **kwargs):
+        self._key = ''
+        self._table = None
+        self._get_sql = None
+        super().__init__(*args, **kwargs)
+
     def set_config(self, config):
         super().set_config(config)
         self._key = self.config.key

setup.cfg

@@ -11,6 +11,14 @@ addopts=
 flake8-max-complexity = 10
 flake8-max-line-length = 99
 
+[mypy]
+ignore_missing_imports = True
+
 [isort]
+not_skip = __init__.py
+known_first_party = aioworkers_pg
+known_third_party = aioworkers
+lines_after_imports = 2
+multi_line_output = 5
 force_single_line = false
 import_heading_localfolder = true

setup.py

@@ -4,37 +4,27 @@
 
 from setuptools import find_packages, setup
 
-version = __import__('aioworkers_pg').__version__
-
-requirements = [
-    'aioworkers>=0.8.0',
-    'asyncpg',
-    'sqlalchemy',
-    'asyncpgsa',
-]
-
-test_requirements = [
-    'pytest',
-    'pytest-runner',
-    'pytest-aioworkers',
-    'pytest-flake8',
-    'flake8-isort',
-]
 
+package = 'aioworkers_pg'
+version = __import__(package).__version__
 readme = pathlib.Path('README.rst').read_text()
 
 
 setup(
     name='aioworkers-pg',
     version=version,
-    description='Module for working with Postgres SQL via asyncpg',
+    description='Module for working with PostgreSQL via asyncpg',
     long_description=readme,
     author='Alexander Bogushov',
     author_email='abogushov@gmail.com',
     url='https://github.com/aioworkers/aioworkers-pg',
-    packages=[i for i in find_packages() if i.startswith('aioworkers_pg')],
+    packages=find_packages(include=[package, package + '.*']),
     include_package_data=True,
-    install_requires=requirements,
+    install_requires=[
+        'aioworkers>=0.15',
+        'asyncpg',
+    ],
+    extras_require={'sa': ['sqlalchemy', 'asyncpgsa']},
     license='Apache Software License 2.0',
     keywords='aioworkers asyncpg',
     classifiers=[
@@ -46,6 +36,4 @@
         'Programming Language :: Python :: 3.6',
         'Programming Language :: Python :: 3.7',
     ],
-    test_suite='tests',
-    tests_require=test_requirements,
 )

tests/test_connector.py

@@ -2,14 +2,14 @@
 
 
 @pytest.fixture
-def config(dsn):
-    from aioworkers.core.config import Config
-    return Config(
+def config(config, dsn):
+    config.update(
         db={
             'cls': 'aioworkers_pg.base.Connector',
             'dsn': dsn,
         },
     )
+    return config
 
 
 async def test_connector(context):

tests/test_pool_close.py

@@ -8,5 +8,6 @@ async def test_pool_close(loop, dsn):
         },
     )
     async with Context(conf, loop=loop) as c:
+        assert c.db._pool is not None
         assert not c.db._pool._closed
-    assert c.db._pool._closed
+    assert c.db._pool is None

tests/test_sa_connector.py

@@ -2,14 +2,14 @@
 
 
 @pytest.fixture
-def config(dsn):
-    from aioworkers.core.config import Config
-    return Config(
+def config(config, dsn):
+    config.update(
         db={
             'cls': 'aioworkers_pg.sa.Connector',
             'dsn': dsn,
         },
     )
+    return config
 
 
 async def test_sa_connector(context):

tests/test_sql.py

@@ -4,14 +4,14 @@
 
 
 @pytest.fixture
-def config(dsn):
-    from aioworkers.core.config import Config
-    return Config(
+def config(config, dsn):
+    config.update(
         db={
             'cls': 'aioworkers_pg.base.Connector',
             'dsn': dsn,
         },
     )
+    return config
 
 
 async def test_select(context, recreate_table):

tests/test_storage.py

@@ -2,9 +2,8 @@
 
 
 @pytest.fixture
-def config(dsn):
-    from aioworkers.core.config import Config
-    return Config(
+def config(config, dsn):
+    config.update(
         db={
             'cls': 'aioworkers_pg.base.Connector',
             'dsn': dsn,
@@ -17,6 +16,7 @@ def config(dsn):
             'format': 'dict',
         },
     )
+    return config
 
 
 async def test_ro_storage(context, recreate_table):