Update to sqlalchemy 1.4b2 (#6446)

* Update code for sqlalchemy 1.4b2

* Bump sqlalchemy version

* Update requirements.txt

* Set dialect

* Update requirements.txt

* Always emit SELECT

frameworks/Python/aiohttp/app/main.py

@@ -24,7 +24,7 @@ from .views import (
 CONNECTION_ORM = os.getenv('CONNECTION', 'ORM').upper() == 'ORM'
 
 
-def pg_dsn() -> str:
+def pg_dsn(dialect=None) -> str:
     """
     :return: DSN url suitable for sqlalchemy and aiopg.
     """
@@ -34,12 +34,11 @@ def pg_dsn() -> str:
         host='tfb-database',
         port='5432',
         username=os.getenv('PGUSER', 'benchmarkdbuser'),
-        drivername='postgresql',
+        drivername='postgresql+{}'.format(dialect) if dialect else 'postgresql',
     ))
 
 
 async def db_ctx(app: web.Application):
-    dsn = pg_dsn()
     # number of gunicorn workers = multiprocessing.cpu_count() as per gunicorn_conf.py
     # max_connections = 2000 as per toolset/setup/linux/databases/postgresql/postgresql.conf:64
     # give 10% leeway
@@ -48,9 +47,11 @@ async def db_ctx(app: web.Application):
     min_size = max(int(max_size / 2), 1)
     print(f'connection pool: min size: {min_size}, max size: {max_size}, orm: {CONNECTION_ORM}')
     if CONNECTION_ORM:
+        dsn = pg_dsn('asyncpg')
         engine = create_async_engine(dsn, future=True, pool_size=max_size)
         app['db_session'] = sessionmaker(engine, class_=AsyncSession)
     else:
+        dsn = pg_dsn()
         app['pg'] = await asyncpg.create_pool(dsn=dsn, min_size=min_size, max_size=max_size, loop=app.loop)
 
     yield

frameworks/Python/aiohttp/app/views.py

@@ -40,9 +40,7 @@ async def single_database_query_orm(request):
     """
     id_ = randint(1, 10000)
     async with request.app['db_session']() as sess:
-        # TODO(SA1.4.0b2): sess.scalar()
-        ret = await sess.execute(select(World.randomnumber).filter_by(id=id_))
-        num = ret.scalar()
+        num = await sess.scalar(select(World.randomnumber).filter_by(id=id_))
     return json_response({'id': id_, 'randomNumber': num})
 
 
@@ -69,9 +67,7 @@ async def multiple_database_queries_orm(request):
     result = []
     async with request.app['db_session']() as sess:
         for id_ in ids:
-            # TODO(SA1.4.0b2): sess.scalar()
-            ret = await sess.execute(select(World.randomnumber).filter_by(id=id_))
-            num = ret.scalar()
+            num = await sess.scalar(select(World.randomnumber).filter_by(id=id_))
             result.append({'id': id_, 'randomNumber': num})
     return json_response(result)
 
@@ -131,17 +127,13 @@ async def updates(request):
     ids = [randint(1, 10000) for _ in range(num_queries)]
     ids.sort()
 
-    # TODO(SA1.4.0b2): async with request.app['db_session'].begin() as sess:
-    async with request.app['db_session']() as sess:
-        async with sess.begin():
-            for id_ in ids:
-                rand_new = randint(1, 10000)
-                # TODO(SA1.4.0b2): world = await sess.get(World, id_)
-                ret = await sess.execute(select(World).filter_by(id=id_))
-                world = ret.scalar()
-                world.randomnumber = rand_new
-
-                result.append({'id': id_, 'randomNumber': rand_new})
+    async with request.app['db_session'].begin() as sess:
+        for id_ in ids:
+            rand_new = randint(1, 10000)
+            world = await sess.get(World, id_, populate_existing=True)
+            world.randomnumber = rand_new
+
+            result.append({'id': id_, 'randomNumber': rand_new})
     return json_response(result)
 
 async def updates_raw(request):

frameworks/Python/aiohttp/requirements.txt

@@ -1,9 +1,9 @@
 aiohttp==3.7.4
-asyncpg==0.21.0
+asyncpg==0.22.0
 cchardet==2.1.7
 gunicorn==20.0.4
 jinja2==2.11.2
 psycopg2==2.8.6
-SQLAlchemy==1.4.0b1
+SQLAlchemy==1.4.0b3
 ujson==2.0.3
 uvloop==0.14.0