initial

Author: jordic

.github/ci.yml

@@ -0,0 +1,59 @@
+name: fastapi_asyncpg
+
+on: [push]
+
+jobs:
+
+  # Job to run pre-checks
+  pre-checks:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.9]
+
+    steps:
+      - name: Checkout the repository
+        uses: actions/checkout@v2
+
+      - name: Setup Python
+        uses: actions/setup-python@v1
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install package
+        run: |
+          pip install flake8==3.7.7
+          pip install mypy==0.720
+          pip install black==19.10b0
+          pip install isort==4.3.21
+      - name: Run pre-checks
+        run: |
+          flake8 fastapi_asyncpg --config=setup.cfg
+          mypy fastapi_asyncpg/ --ignore-missing-imports
+          isort -c -rc fastapi_asyncpg/
+          black -l 80 --check --verbose fastapi_asyncpg
+  # Job to run tests
+  tests:
+    runs-on: ubuntu-latest
+    # Set environment variables
+    steps:
+      - name: Checkout the repository
+        uses: actions/checkout@v2
+
+      - name: Setup Python
+        uses: actions/setup-python@v1
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install the package
+        run: |
+          pip install -e .[test]
+
+      - name: Run tests
+        run: |
+          pytest -rfE --cov=fastapi_asyncpg -s --tb=native -v --cov-report xml --cov-append tests
+
+      - name: Upload coverage to Codecov
+        uses: codecov/codecov-action@v1
+        with:
+          file: ./coverage.xml

.gitignore

@@ -0,0 +1,21 @@
+**/__pycache__
+env/
+venv/
+.coverage
+.env
+.idea
+.installed.cfg
+.pytest_cache/
+.mypy_cache/
+.tox/
+bin/
+coverage.xml
+develop-eggs/
+lib/
+lib64
+parts/
+pyvenv.cfg
+*.egg-info
+*.profraw
+*.py?
+*.swp

.isort.cfg

@@ -0,0 +1,5 @@
+[settings]
+force_single_line=True
+sections=FUTURE,THIRDPARTY,FIRSTPARTY,LOCALFOLDER,STDLIB
+no_lines_before=LOCALFOLDER,THIRDPARTY,FIRSTPARTY,STDLIB
+force_alphabetical_sort=True

LICENSE

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2020 Jordi collell
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Makefile

@@ -0,0 +1,15 @@
+.PHONY: isort black flake8 mypy
+
+lint: isort black flake8 mypy
+
+isort:
+	isort fastapi_asyncpg
+
+black:
+	black fastapi_asyncpg/  -l 80
+
+flake8:
+	flake8 fastapi_asyncpg
+
+mypy:
+	mypy fastapi_asyncpg

README.md

@@ -0,0 +1,5 @@
+# FastAPI AsyncPG
+
+## Authors
+
+`fastapi_asyncpg` was written by `Jordi collell <jordic@gmail.com>`\_.

examples/__init__.py

@@ -0,0 +1,61 @@
+from fastapi import FastAPI
+from fastapi import Depends
+from fastapi_asyncpg import configure_asyncpg
+
+import pydantic as pd
+
+app = FastAPI()
+
+
+db = configure_asyncpg(app, "postgresql://postgres:postgres@localhost/db")
+
+
+class Demo(pd.BaseModel):
+    key: str
+    value: str
+
+
+class DemoObj(Demo):
+    demo_id: int
+
+
+@db.on_init
+async def initialize_db(db):
+    await db.execute(
+        """
+        CREATE TABLE IF NOT EXISTS demo (
+            demo_id serial primary key,
+            key varchar not null,
+            value varchar not null,
+            UNIQUE(key)
+        );
+    """
+    )
+
+
+@app.post("/", response_model=DemoObj)
+async def add_resource(data: Demo, db=Depends(db.connection)):
+    """
+    Add a resource to db:
+    curl -X POST -d '{"key": "test", "value": "asdf"}' \
+        http://localhost:8000/
+    """
+    result = await db.fetchrow(
+        """
+        INSERT into demo values (default, $1, $2) returning *
+    """,
+        data.key,
+        data.value,
+    )
+    return dict(result)
+
+
+@app.get("/{key:str}", response_model=DemoObj)
+async def get_resouce(key: str, db=Depends(db.connection)):
+    result = await db.fetchrow(
+        """
+        SELECT * from demo where key=$1
+    """,
+        key,
+    )
+    return dict(result)

fastapi_asyncpg/__init__.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+from fastapi import FastAPI
+
+import asyncpg
+import typing
+
+
+async def noop(db: asyncpg.Connection):
+    return
+
+
+class configure_asyncpg:
+    def __init__(
+        self,
+        app: FastAPI,
+        dsn: str,
+        *,
+        init_db: typing.Callable = None,  # callable for running sql on init
+        pool=None,  # usable on testing
+        **options,
+    ):
+        self.app = app
+        self.dsn = dsn
+        self.init_db = init_db
+        self.con_opts = options
+        self._pool = pool
+        self.app.router.add_event_handler("startup", self.on_connect)
+        self.app.router.add_event_handler("shutdown", self.on_disconnect)
+
+    async def on_connect(self):
+        if self._pool:
+            self.app.state.pool = self._pool
+            return
+        pool = await asyncpg.create_pool(dsn=self.dsn, **self.con_opts)
+        async with pool.acquire() as db:
+            await self.init_db(db)
+        self.app.state.pool = pool
+
+    async def on_disconnect(self):
+        await self.app.state.pool.close()
+
+    def on_init(self, func):
+        self.init_db = func
+        return func
+
+    @property
+    def pool(self):
+        return self.app.state.pool
+
+    async def connection(self):
+        async with self.pool.acquire() as db:
+            yield db
+
+    async def transaction(self):
+        async with self.pool.acquire() as db:
+            txn = db.transaction()
+            await txn.start()
+            try:
+                yield db
+            except:
+                await txn.rollback()
+                raise
+            else:
+                await txn.commit()
+
+    atomic = transaction
+
+
+class SingleConnectionTestingPool:
+    """A fake pool that simulates pooling, but runs on
+    a single transaction that it's rolled back after
+    each test.
+    With some large schemas this seems to be faster than
+    the other approach
+    """
+
+    def __init__(
+        self,
+        conn: asyncpg.Connection,
+        initialize: typing.Callable = None,
+        add_logger_postgres: bool = False,
+    ):
+        self._conn = conn
+        self.tx = None
+        self.started = False
+        self.add_logger_postgres = add_logger_postgres
+        self.initialize = initialize
+
+    def acquire(self, *, timeout=None):
+        return ConAcquireContext(self._conn, self)
+
+    async def start(self):
+        if self.started:
+            return
+
+        def log_postgresql(con, message):
+            print(message)
+
+        if self.add_logger_postgres:
+            self._conn.add_log_listener(log_postgresql)
+        self.tx = self._conn.transaction()
+        await self.tx.start()
+        await self.initialize(self._conn)
+        self.started = True
+
+    async def release(self):
+        if self.tx:
+            await self.tx.rollback()
+
+    def __getattr__(self, key):
+        return getattr(self._conn, key)
+
+
+async def create_pool_test(
+    dsn: str,
+    *,
+    initialize: typing.Callable = None,
+    add_logger_postgres: bool = False,
+):
+    """This part is only used for testing,
+    we create a fake "pool" that just starts a connecion,
+    that does a transaction inside it"""
+    conn = await asyncpg.connect(dsn=dsn)
+    pool = SingleConnectionTestingPool(
+        conn, initialize=initialize, add_logger_postgres=add_logger_postgres
+    )
+    return pool
+
+
+class ConAcquireContext:
+    def __init__(self, conn, manager):
+        self._conn = conn
+        self.manager = manager
+
+    async def __aenter__(self):
+        if not self.manager.tx:
+            await self.manager.start()
+        self.tr = self._conn.transaction()
+        await self.tr.start()
+        return self._conn
+
+    async def __aexit__(self, exc_type, exc, tb):
+        if exc_type:
+            await self.tr.rollback()
+        else:
+            await self.tr.commit()