try fix pypy

Author: bymoye

frameworks/Python/blacksheep/app-socketify.py

@@ -1,27 +1,22 @@
 import multiprocessing
 import os
-import asyncpg
+import psycopg
 import platform
 import random
 import asyncio
 import blacksheep as bs
 import jinja2
-# import json
 from pathlib import Path
-try:
-    import uvloop
-    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
-except Exception:
-    ...
-
-READ_ROW_SQL = 'SELECT "id", "randomnumber" FROM "world" WHERE id = $1'
-WRITE_ROW_SQL = 'UPDATE "world" SET "randomnumber"=$1 WHERE id=$2'
+from psycopg_pool import AsyncConnectionPool
+
+READ_ROW_SQL = 'SELECT "id", "randomnumber" FROM "world" WHERE id = %s'
+WRITE_ROW_SQL = 'UPDATE "world" SET "randomnumber"=%s WHERE id=%s'
 ADDITIONAL_ROW = [0, "Additional fortune added at request time."]
 MAX_CONNECTIONS = 1900
 CORE_COUNT = multiprocessing.cpu_count()
 PROCESSES = CORE_COUNT
-MAX_POOL_SIZE = max(1,int(os.getenv('MAX_POOL_SIZE', MAX_CONNECTIONS // PROCESSES)))
-MIN_POOL_SIZE = max(1,int(os.getenv('MIN_POOL_SIZE', MAX_POOL_SIZE // 2)))
+MAX_POOL_SIZE = max(1, int(os.getenv('MAX_POOL_SIZE', MAX_CONNECTIONS // PROCESSES)))
+MIN_POOL_SIZE = max(1, int(os.getenv('MIN_POOL_SIZE', MAX_POOL_SIZE // 2)))
 
 WORKER_PROCESSES = CORE_COUNT
 MAX_POOL_SIZE = max(1, MAX_CONNECTIONS // WORKER_PROCESSES)
@@ -30,18 +25,19 @@
 
 async def setup_db(app):
     global db_pool
-    db_pool = await asyncpg.create_pool(
-        user=os.getenv('PGUSER', "benchmarkdbuser"),
-        password=os.getenv('PGPASS', "benchmarkdbpass"),
-        database='hello_world',
-        host="tfb-database",
-        port=5432,
+    conninfo = (
+        f"postgresql://{os.getenv('PGUSER', 'benchmarkdbuser')}:{os.getenv('PGPASS', 'benchmarkdbpass')}"
+        f"@tfb-database:5432/hello_world"
+    )
+    db_pool = AsyncConnectionPool(
+        conninfo=conninfo,
         min_size=MIN_POOL_SIZE,
         max_size=MAX_POOL_SIZE,
+        open=False,
     )
+    await db_pool.open()
 
 async def shutdown_db(app):
-    """Close asyncpg connection pool for the current process."""
     global db_pool
     if db_pool is not None:
         await db_pool.close()
@@ -51,7 +47,6 @@ def load_fortunes_template():
     with Path("templates/fortune.html").open("r") as f:
         return jinja2.Template(f.read())
 
-
 fortune_template = load_fortunes_template()
 
 app = bs.Application()
@@ -70,76 +65,62 @@ def get_num_queries(request):
 
 JSON_CONTENT_TYPE = b"application/json"
 
-
-# ------------------------------------------------------------------------------------------
-
 @bs.get('/json')
 async def json_test(request):
-    return bs.json( {'message': 'Hello, world!'} )
+    return bs.json({'message': 'Hello, world!'})
 
 @bs.get('/db')
 async def single_db_query_test(request):
     row_id = random.randint(1, 10000)
-    
-    async with db_pool.acquire() as db_conn:
-        number = await db_conn.fetchval(READ_ROW_SQL, row_id)
-    
-    # return jsonify(Result(id=row_id, randomNumber=number))
-    # return ({'id': row_id, 'randomNumber': number})
-    return bs.json({'id': row_id, 'randomNumber': number})
-
+    async with db_pool.connection() as db_conn:
+        async with db_conn.cursor() as cursor:
+            await cursor.execute(READ_ROW_SQL, (row_id,))
+            number = await cursor.fetchone()
+    return bs.json({'id': row_id, 'randomNumber': number[1]})
 
 @bs.get('/queries')
 async def multiple_db_queries_test(request):
     num_queries = get_num_queries(request)
     row_ids = random.sample(range(1, 10000), num_queries)
     worlds = []
-
-    async with db_pool.acquire() as db_conn:
-        statement = await db_conn.prepare(READ_ROW_SQL)
-        for row_id in row_ids:
-            number = await statement.fetchval(row_id)
-            worlds.append( {"id": row_id, "randomNumber": number} )
-            # worlds.append(Result(id=row_id, randomNumber=number))
-
+    async with db_pool.connection() as db_conn:
+        async with db_conn.cursor() as cursor:
+            for row_id in row_ids:
+                await cursor.execute(READ_ROW_SQL, (row_id,))
+                number = await cursor.fetchone()
+                worlds.append({"id": row_id, "randomNumber": number[1]})
     return bs.json(worlds)
-    # return jsonify(worlds)
-
 
 @bs.get('/fortunes')
 async def fortunes_test(request):
-    async with db_pool.acquire() as db_conn:
-        fortunes = await db_conn.fetch("SELECT * FROM Fortune")
-
+    async with db_pool.connection() as db_conn:
+        async with db_conn.cursor() as cursor:
+            await cursor.execute("SELECT * FROM Fortune")
+            fortunes = await cursor.fetchall()
     fortunes.append(ADDITIONAL_ROW)
     fortunes.sort(key=lambda row: row[1])
     data = fortune_template.render(fortunes=fortunes)
     return bs.html(data)
 
-
 @bs.get('/updates')
 async def db_updates_test(request):
     num_queries = get_num_queries(request)
     updates = list(zip(
         random.sample(range(1, 10000), num_queries),
         sorted(random.sample(range(1, 10000), num_queries))
     ))
-    # worlds = [Result(id=row_id, randomNumber=number) for row_id, number in updates]
-    worlds = [ {"id": row_id, "randomNumber": number} for row_id, number in updates ]
-    async with db_pool.acquire() as db_conn:
-        statement = await db_conn.prepare(READ_ROW_SQL)
-        for row_id, _ in updates:
-            await statement.fetchval(row_id)
-        await db_conn.executemany(WRITE_ROW_SQL, updates)
+    worlds = [{"id": row_id, "randomNumber": number} for row_id, number in updates]
+    async with db_pool.connection() as db_conn:
+        async with db_conn.cursor() as cursor:
+            for row_id, number in updates:
+                await cursor.execute(READ_ROW_SQL, (row_id,))
+                await cursor.fetchone()
+            await cursor.executemany(WRITE_ROW_SQL, [(number, row_id) for row_id, number in updates])
     return bs.json(worlds)
-    # return jsonify(worlds)
-
 
 @bs.get('/plaintext')
 async def plaintext_test(request):
     return bs.Response(200, content=bs.Content(b"text/plain", b'Hello, World!'))
-    #return bs.text('Hello, World!')
-
 
 if platform.python_implementation() == 'PyPy':
     from socketify import ASGI
@@ -150,16 +131,11 @@ async def plaintext_test(request):
     def run_app():
         ASGI(app).listen(8080, lambda config: logging.info(f"Listening on port http://localhost:{config.port} now\n")).run()
 
-
     def create_fork():
         n = os.fork()
-        # n greater than 0 means parent process
         if not n > 0:
             run_app()
 
-
-    # fork limiting the cpu count - 1
     for i in range(1, workers):
         create_fork()
-
     run_app()

frameworks/Python/blacksheep/blacksheep-socketify-asgi-pypy.dockerfile

@@ -4,7 +4,7 @@ ADD ./ /blacksheep
 
 WORKDIR /blacksheep
 
-RUN apt-get update; apt-get install libuv1 -y
+RUN apt-get update; apt-get install libuv1 libpq5 -y
 
 RUN pip3 install -r /blacksheep/requirements.txt
 RUN pip3 install -r /blacksheep/requirements-pypy.txt

frameworks/Python/blacksheep/requirements-pypy.txt

@@ -1,2 +1,3 @@
+psycopg[pool]
 socketify
 h11

frameworks/Python/blacksheep/requirements-uvicorn.txt

@@ -3,3 +3,4 @@ uvicorn==0.34.1
 httptools==0.6.4
 gunicorn==23.0.0
 msgspec==0.19.0
+asyncpg==0.30.0

frameworks/Python/blacksheep/requirements.txt

@@ -1,3 +1,3 @@
-asyncpg==0.30.0
 Jinja2==3.1.6
 blacksheep==2.3.1
+psycopg