Updates

backend/pyproject.toml

@@ -33,8 +33,9 @@ fastapi = "^0.112.0"
 mypy = "^1.11.2"
 isort = "^5.13.2"
 pydantic-settings = "^2.4.0"
-uvicorn = "^0.30.6"
+uvicorn = {extras = ["standard"], version = "^0.31.0"}
 asyncpg = "^0.29.0"
+uvloop = "^0.20.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.3.1"

backend/startup.py

@@ -82,4 +82,4 @@ async def dispatch(self, request: Request, call_next):
     import uvicorn
 
     # uvicorn.run(fastapi_app, host="0.0.0.0", port=5000)
-    uvicorn.run(fastapi_app, port=5000)
+    uvicorn.run(fastapi_app, port=5000)

backend/v2/allocations/repositories.py

@@ -132,26 +132,56 @@ async def get_last_allocation_request_nonce(
 ) -> int | None:
     """Get the last nonce of the allocation requests for a user."""
 
-    result = await session.execute(
-        select(AllocationRequestDB.nonce).
-        join(User, AllocationRequestDB.user_id == User.id).
-        filter(User.address == user_address).
-        order_by(AllocationRequestDB.nonce.desc()).
-        limit(1)
-    )
-    return result.scalar()
+    import time
+
+
+
+    # return result.scalar()
+
+    start = time.time()
 
     user = await get_user_by_address(session, user_address)
     if user is None:
         return None
 
-    result = await session.execute(
+    # result = await session.execute(
+    #     select(func.max(AllocationRequestDB.nonce)).filter(
+    #         AllocationRequestDB.user_id == user.id
+    #     )
+    # )
+
+    print("get_last_allocation_request_nonce2", time.time() - start)
+
+    start = time.time()
+
+    result = await session.scalar(
         select(func.max(AllocationRequestDB.nonce)).filter(
             AllocationRequestDB.user_id == user.id
         )
     )
 
-    return result.scalar()
+    # result = await session.execute(
+    #     select(AllocationRequestDB.nonce).
+    #     join(User, AllocationRequestDB.user_id == User.id).
+    #     filter(User.address == user_address).
+    #     order_by(AllocationRequestDB.nonce.desc()).
+    #     limit(1)
+    # )
+
+    print("get_last_allocation_request_nonce", time.time() - start)
+
+    # start = time.time()
+
+    # result = (
+    #     AllocationRequestDB.query.join(User, User.id == AllocationRequestDB.user_id)
+    #     .filter(User.address == user_address)
+    #     .order_by(AllocationRequestDB.nonce.desc())
+    #     .first()
+    # )
+
+    # print("?????????get_user_last_allocation_request", time.time() - start)
+
+    return result
 
 
 async def get_donations_by_project(

backend/v2/allocations/router.py

@@ -20,6 +20,10 @@ async def allocate(
     Make an allocation for the user.
     """
 
+    import time
+
+
+    start = time.time()
     request = UserAllocationRequest(
         user_address=allocation_request.user_address,
         allocations=allocation_request.payload.allocations,
@@ -36,4 +40,6 @@ async def allocate(
     print("pending_epoch", pending_epoch)
 
     await allocations.make(pending_epoch, request)
+
+    print("allocate took: ", time.time() - start)
 

backend/v2/allocations/services.py

@@ -1,3 +1,4 @@
+import asyncio
 from dataclasses import dataclass
 import time
 
@@ -158,12 +159,18 @@ async def simulate_leverage(
 
     start_time = time.time()
 
-    all_projects = await projects.get_project_addresses(epoch_number)
+    all_projects, matched_rewards, existing_allocations = await asyncio.gather(
+        projects.get_project_addresses(epoch_number),
+        estimated_project_matched_rewards.get(epoch_number),
+        get_allocations_with_user_uqs(session, epoch_number),
+    )
+
+    # all_projects = await projects.get_project_addresses(epoch_number)
 
-    matched_rewards = await estimated_project_matched_rewards.get(epoch_number)
+    # matched_rewards = await estimated_project_matched_rewards.get(epoch_number)
 
-    # Get all allocations before user's allocation
-    existing_allocations = await get_allocations_with_user_uqs(session, epoch_number)
+    # # Get all allocations before user's allocation
+    # existing_allocations = await get_allocations_with_user_uqs(session, epoch_number)
 
     print("existing allocations retrieved in", time.time() - start_time)
 

backend/v2/allocations/validators.py

@@ -1,3 +1,4 @@
+import asyncio
 from dataclasses import dataclass
 from web3 import AsyncWeb3
 from app import exceptions
@@ -27,21 +28,20 @@ async def verify(self, epoch_number: int, request: UserAllocationRequest) -> Non
         import time
         start = time.time()
 
-        await verify_logic(
-            session=self.session,
-            epoch_subgraph=self.epochs_subgraph,
-            projects_contracts=self.projects_contracts,
-            epoch_number=epoch_number,
-            payload=request,
-        )
-
-        print("verify_logic", time.time() - start)
-
-        await verify_signature(
-            w3=self.projects_contracts.w3,
-            chain_id=self.chain_id,
-            user_address=request.user_address,
-            payload=request,
+        await asyncio.gather(
+            verify_logic(
+                session=self.session,
+                epoch_subgraph=self.epochs_subgraph,
+                projects_contracts=self.projects_contracts,
+                epoch_number=epoch_number,
+                payload=request,
+            ),
+            verify_signature(
+                w3=self.projects_contracts.w3,
+                chain_id=self.chain_id,
+                user_address=request.user_address,
+                payload=request,
+            ),
         )
 
         print("verify_signature", time.time() - start)
@@ -61,47 +61,141 @@ async def verify_logic(
     # if epoch_details.state != "PENDING":
     #     raise exceptions.NotInDecision
 
-    import time
-
     # Check if the allocations are not empty
     if not payload.allocations:
         raise exceptions.EmptyAllocations()
 
-    start = time.time()    
+    print("already here")
+
+    async def _check_database():
+        await _provided_nonce_matches_expected(session, payload.user_address, payload.nonce)
+        await _user_is_not_patron(session, epoch_subgraph, payload.user_address, epoch_number)
+        await _user_has_budget(session, payload, epoch_number)
+
+    await asyncio.gather(
+        _check_database(),
+        _provided_projects_are_correct(projects_contracts, epoch_number, payload)
+    )
+
+
+    # try:
+    #     async with asyncio.TaskGroup() as tg:
+
+    #         tg.create_task(_provided_nonce_matches_expected(session, payload.user_address, payload.nonce))
+    #         tg.create_task(_user_is_not_patron(session, epoch_subgraph, payload.user_address, epoch_number))
+    #         tg.create_task(_provided_projects_are_correct(projects_contracts, epoch_number, payload))
+    #         tg.create_task(_user_has_budget(session, payload, epoch_number))
+    # except Exception as e:
+    #     print("e", e)
+    #     raise e
+
+
+    # summary = asyncio.gather(
+    #     _provided_nonce_matches_expected(session, payload.user_address, payload.nonce),
+    #     _user_is_not_patron(
+    #         session, epoch_subgraph, payload.user_address, epoch_number
+    #     ),
+    #     _provided_projects_are_correct(
+    #         projects_contracts, epoch_number, payload
+    #     ),
+    #     _user_has_budget(session, payload, epoch_number),
+    #     return_exceptions=True,
+    # )
+
+    # print("maybe here?")
+
+    # for i in await summary:
+    #     if isinstance(i, Exception):
+    #         raise i
+
+    print("hehehehehe")
+
+async def _provided_nonce_matches_expected(
+    # Component dependencies
+    session: AsyncSession,
+    # Arguments
+    user_address: str,
+    nonce: int,
+) -> None:
+    """
+    Check if the nonce is as expected.
+    """
+    # Get the next nonce
+    next_nonce = await get_next_user_nonce(session, user_address)
+
     # Check if the nonce is as expected
-    expected_nonce = await get_next_user_nonce(session, payload.user_address)
-    if payload.nonce != expected_nonce:
-        raise exceptions.WrongAllocationsNonce(payload.nonce, expected_nonce)
+    if nonce != next_nonce:
+        raise exceptions.WrongAllocationsNonce(nonce, next_nonce)
 
-    print("get_next_user_nonce", time.time() - start)   
 
-    start = time.time()
+async def _user_is_not_patron(
+    # Component dependencies
+    session: AsyncSession,
+    epoch_subgraph: EpochsSubgraph,
+    # Arguments
+    user_address: str,
+    epoch_number: int,
+) -> None:
+    """
+    Check if the user is not a patron.
+    """
     # Check if the user is not a patron
     epoch_details = await epoch_subgraph.get_epoch_by_number(epoch_number)
     is_patron = await user_is_patron_with_budget(
         session,
-        payload.user_address,
+        user_address,
         epoch_number,
         epoch_details.finalized_timestamp.datetime(),
     )
     if is_patron:
-        raise exceptions.NotAllowedInPatronMode(payload.user_address)
+        raise exceptions.NotAllowedInPatronMode(user_address)
 
-    print("user_is_patron_with_budget", time.time() - start)
+async def get_next_user_nonce(
+    # Component dependencies
+    session: AsyncSession,
+    # Arguments
+    user_address: str,
+) -> int:
+    """
+    Get the next expected nonce for the user.
+    It's a simple increment of the last nonce, or 0 if there is no previous nonce.
+    """
+    # Get the last allocation request of the user
+    last_allocation_request = await get_last_allocation_request_nonce(
+        session, user_address
+    )
 
+    print("last_allocation_request", last_allocation_request)
 
-    start = time.time()
+    # Calculate the next nonce
+    if last_allocation_request is None:
+        return 0
 
+    # Increment the last nonce
+    return last_allocation_request + 1
+
+
+async def _provided_projects_are_correct(
+    # Component dependencies
+    projects_contracts: ProjectsContracts,
+    # Arguments
+    epoch_number: int,
+    payload: UserAllocationRequest,
+) -> None:
+    """
+    Check if the projects in the allocation request are correct.
+    """
+
+    import time
+    start = time.time()
     # Check if the user is not a project
     all_projects = await projects_contracts.get_project_addresses(epoch_number)
     if payload.user_address in all_projects:
         raise exceptions.ProjectAllocationToSelf()
 
-    project_addresses = [a.project_address for a in payload.allocations]
-
     print("get_project_addresses", time.time() - start)
 
-    start = time.time()
+    project_addresses = [a.project_address for a in payload.allocations]
 
     # Check if the projects are valid
     invalid_projects = set(project_addresses) - set(all_projects)
@@ -113,10 +207,19 @@ async def verify_logic(
     if duplicates:
         raise exceptions.DuplicatedProjects(duplicates)
 
-    print("invalid_projects", time.time() - start)
 
+async def _user_has_budget(
+    # Component dependencies
+    session: AsyncSession,
+    # Arguments
+    payload: UserAllocationRequest,
+    epoch_number: int,
+) -> None:
+    """
+    Check if the user has enough budget for the allocation.
+    Check if the sum of the allocations is within the user's budget.
+    """
 
-    start = time.time()
     # Get the user's budget
     user_budget = await get_budget_by_user_address_and_epoch(
         session, payload.user_address, epoch_number
@@ -129,32 +232,6 @@ async def verify_logic(
     if sum(a.amount for a in payload.allocations) > user_budget:
         raise exceptions.RewardsBudgetExceeded()
 
-    print("get_budget_by_user_address_and_epoch", time.time() - start)
-
-async def get_next_user_nonce(
-    # Component dependencies
-    session: AsyncSession,
-    # Arguments
-    user_address: str,
-) -> int:
-    """
-    Get the next expected nonce for the user.
-    It's a simple increment of the last nonce, or 0 if there is no previous nonce.
-    """
-    # Get the last allocation request of the user
-    last_allocation_request = await get_last_allocation_request_nonce(
-        session, user_address
-    )
-
-    print("last_allocation_request", last_allocation_request)
-
-    # Calculate the next nonce
-    if last_allocation_request is None:
-        return 0
-
-    # Increment the last nonce
-    return last_allocation_request + 1
-
 
 async def verify_signature(
     w3: AsyncWeb3, chain_id: int, user_address: str, payload: UserAllocationRequest

backend/v2/core/dependencies.py

@@ -1,18 +1,20 @@
+from asyncio import current_task
 from contextlib import asynccontextmanager
+from functools import lru_cache
 from typing import Annotated, AsyncGenerator
 
 from fastapi import Depends
 from app.infrastructure.database.models import BaseModel
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine, async_scoped_session
 from web3 import AsyncHTTPProvider, AsyncWeb3
 from web3.middleware import async_geth_poa_middleware
 
 
 class OctantSettings(BaseSettings):
-    model_config = SettingsConfigDict(env_file='.env', extra='ignore')
+    model_config = SettingsConfigDict(env_file='.env', extra='ignore', frozen=True)
 
 
 class Web3ProviderSettings(OctantSettings):
@@ -55,18 +57,14 @@ async def create_tables():
         await conn.run_sync(BaseModel.metadata.create_all)
 
 
-# @asynccontextmanager
-async def get_db_session(
+@lru_cache(1)
+def get_sessionmaker(
     settings: Annotated[DatabaseSettings, Depends(get_database_settings)]
-) -> AsyncGenerator[AsyncSession, None]:
-    # Create an async SQLAlchemy engine
-
-    # logging.error("Creating database engine")
-
+) -> async_sessionmaker[AsyncSession]:
     engine = create_async_engine(
         settings.sqlalchemy_database_uri,
         echo=False,                    # Disable SQL query logging (for performance)
-        pool_size=20,                  # Initial pool size (default is 5)
+        pool_size=100,                 # Initial pool size (default is 5)
         max_overflow=10,               # Extra connections if pool is exhausted
         pool_timeout=30,               # Timeout before giving up on a connection
         pool_recycle=3600,             # Recycle connections after 1 hour (for long-lived connections)
@@ -75,22 +73,52 @@ async def get_db_session(
         # connect_args={"options": "-c timezone=utc"}  # Ensures timezone is UTC
     )
 
-    # Create a sessionmaker with AsyncSession class
-    async_session = async_sessionmaker(
+    sessionmaker = async_sessionmaker(
         autocommit=False, autoflush=False, bind=engine, class_=AsyncSession
     )
 
+    scoped_session = async_scoped_session(sessionmaker, scopefunc=current_task)
+
+    return scoped_session
+
+# @asynccontextmanager
+async def get_db_session(
+    sessionmaker: Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)]
+) -> AsyncGenerator[AsyncSession, None]:
+    # Create an async SQLAlchemy engine
+
+    # logging.error("Creating database engine")
+
+    # engine = create_async_engine(
+    #     settings.sqlalchemy_database_uri,
+    #     echo=False,                    # Disable SQL query logging (for performance)
+    #     pool_size=20,                  # Initial pool size (default is 5)
+    #     max_overflow=10,               # Extra connections if pool is exhausted
+    #     pool_timeout=30,               # Timeout before giving up on a connection
+    #     pool_recycle=3600,             # Recycle connections after 1 hour (for long-lived connections)
+    #     pool_pre_ping=True,            # Check if the connection is alive before using it
+    #     future=True,                   # Use the future-facing SQLAlchemy 2.0 style
+    #     # connect_args={"options": "-c timezone=utc"}  # Ensures timezone is UTC
+    # )
+
+    # # Create a sessionmaker with AsyncSession class
+    # async_session = async_sessionmaker(
+    #     autocommit=False, autoflush=False, bind=engine, class_=AsyncSession
+    # )
+
     # logging.error("Opening session", async_session)
 
     # Create a new session
-    async with async_session() as session:
+    async with sessionmaker() as session:
         try:
             yield session
             await session.commit()
-        except Exception:
+        except Exception as e:
+            print("----Rolling back session, error:", e)
             await session.rollback()
             raise
         finally:
+            print("----Closing session")
             await session.close()
 
 

backend/v2/uniqueness_quotients/repositories.py

@@ -50,12 +50,18 @@ async def get_gp_stamps_by_address(
 ) -> GPStamps | None:
     """Gets the latest GitcoinPassport Stamps record for a user."""
 
-    result = await session.execute(
-        select(GPStamps)
-        .join(User)
-        .filter(User.address == to_checksum_address(user_address))
-        .order_by(GPStamps.created_at.desc())
-        .limit(1)
+    user = await get_user_by_address(session, user_address)
+
+    result = await session.scalar(
+        select(GPStamps).filter(GPStamps.user_id == user.id).order_by(GPStamps.created_at.desc()).limit(1)
     )
 
-    return result.scalar_one_or_none()
+    # result = await session.execute(
+    #     select(GPStamps)
+    #     .join(User)
+    #     .filter(User.address == to_checksum_address(user_address))
+    #     .order_by(GPStamps.created_at.desc())
+    #     .limit(1)
+    # )
+
+    return result

backend/v2/users/repositories.py

@@ -7,5 +7,15 @@
 async def get_user_by_address(session: AsyncSession, user_address: str) -> User | None:
     user_address = to_checksum_address(user_address)
 
-    result = await session.execute(select(User).filter(User.address == user_address))
-    return result.scalar_one_or_none()
+    import time
+    start = time.time()
+
+    result = await session.scalar(
+        select(User).filter(User.address == user_address).limit(1)
+    )
+
+    # result = await session.execute(select(User).filter(User.address == user_address))
+    print("get_user_by_address", time.time() - start)
+    print("result", result)
+
+    return result

backend/ws_allocation_tester.py

@@ -95,7 +95,7 @@ async def allocate():
     try:
         # url = f"https://uat-backend.octant.wildland.dev/allocations/users/{me.address}/allocation_nonce"
         # url = f"https://master-backend.octant.wildland.dev/allocations/users/{me.address}/allocation_nonce"
-        url = f"http://127.0.0.1:5000/allocations/users/{me.address}/allocation_nonce"
+        url = f"http://127.0.0.1:5000/allocations/users/{me.address}/allocation_nonce"  # forward to flask
         # url = f"http://127.0.0.1:5000/flask/allocations/users/{me.address}/allocation_nonce"
         nonce = requests.get(url).json()['allocationNonce']
 
@@ -110,14 +110,15 @@ async def allocate():
     sig_time = time.time()
 
     random_mult = random.random()
-    amout = int(12223333 * random_mult)
+    amout = int(1222333312223333 * random_mult)
+    # amout = 827243882781739
     # print("Amount: ", amout)    
 
     payload = {
         "allocations": [
             {
                 "proposalAddress": address,
-                "amount": amout,
+                "amount": str(amout),
             }
             for address in project_addresses
         ],
@@ -149,7 +150,7 @@ async def allocate():
     resp = requests.post(
         # "https://uat-backend.octant.wildland.dev/allocations/allocate",
         # "https://master-backend.octant.wildland.dev/allocations/allocate",
-        "http://127.0.0.1:5000/allocations/allocate",
+        "http://127.0.0.1:5000/allocations/allocate", # async fapi
         # "http://127.0.0.1:5000/flask/allocations/allocate",
         json=request_data,
     )
@@ -168,7 +169,7 @@ async def allocate():
 
 async def run_ws():
 
-    for i in range(10):
+    for i in range(5):
 
         global pre_allocate
         global donors_count

backend/ws_req_metrics.py

@@ -127,7 +127,7 @@ def process_file(file_path):
         return None
 
 def main():
-    ws_logs_dir = 'wr_logs_flask_1'
+    ws_logs_dir = 'wr_logs'
     if not os.path.exists(ws_logs_dir):
         print(f"Directory {ws_logs_dir} does not exist.")
         return