Refactor: Use Async DB instead of Sync #817
Open
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
1yam
commented
Jul 8, 2025
There was a problem hiding this comment.
would make sense that @amalcaraz review this parts (mostly the update part)
1yam
force-pushed
the
1yam-sync-db-to-async
branch
from
f113cd1 to
64c1b98
Compare
nesitor
reviewed
Jul 17, 2025
Comment on lines
+81
to
84
| select( | ||
| AggregateDb.key, | ||
| AggregateDb.content, | ||
| AggregateDb.creation_datetime.label("created"), |
There was a problem hiding this comment.
We need to ensure that this is the same equivalent
Comment on lines
+97
to
+101
| select(AggregateDb.key, AggregateDb.content) | ||
| .filter(where_clause) | ||
| .order_by(AggregateDb.key) | ||
| ) | ||
| result = query.all() | ||
| result = (await session.execute(query)).all() |
Comment on lines
+21
to
+23
| if dapp: | ||
| # For some reason asyncpg don't handle it if dapp is None | ||
| query = query.where(AlephBalanceDb.dapp == dapp) |
There was a problem hiding this comment.
Suggested change
| if dapp: | |
| # For some reason asyncpg don't handle it if dapp is None | |
| query = query.where(AlephBalanceDb.dapp == dapp) | |
| if dapp: | |
| # For some reason asyncpg don't handle it if dapp is None | |
| query = query.where(AlephBalanceDb.dapp == dapp) | |
| else: | |
| query = query.where(AlephBalanceDb.dapp is null) |
This is not correct, as if dapp is None don't filter it, we need to equal to None or Null in DB.
Comment on lines
+268
to
+269
| select_stmt += " WHERE address = ANY(:addresses)" | ||
| parameters = {"addresses": list(addresses)} |
There was a problem hiding this comment.
I'm not sure if this is totally equivalent, maybe we need to maintain IN operator?
|
|
||
| return { | ||
| column.name: getattr(self, column.name) | ||
| for column in self.__table__.columns | ||
| if column.name not in exclude_set | ||
| if column.name not in exclude_set and column.name not in insp.unloaded |
| @@ -49,9 +49,14 @@ def update_balances(session: DbSession, content: Mapping[str, Any]) -> None: | |||
| dapp = content.get("dapp") | |||
|
|
|||
| LOGGER.info("Updating balances for %s (dapp: %s)", chain, dapp) | |||
| print(f"Chain type: {type(chain)}, Chain value: {chain.value}, Full chain: {chain}") | |||
Comment on lines
+55
to
+57
| print(f"Number of balances to update: {len(balances)}") | ||
| for addr, bal in list(balances.items())[:3]: # Print first 3 for debug | ||
| print(f" {addr}: {bal} bal type {type(bal)}") |
There was a problem hiding this comment.
Also, these one as are debug messages I think
| return None | ||
|
|
||
| # TODO: fix this pydanticV2 issue |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
To bring back message parallelization we need to use AsyncDb instead of sync one
Related Clickup or Jira tickets : ALEPH-526
Self proofreading checklist
Changes
This pull request updates the database session handling across multiple files to use asynchronous methods and factories, improving compatibility with modern async workflows. The changes primarily involve replacing synchronous session handling with asynchronous equivalents and updating method calls accordingly.
Transition to Asynchronous Database Sessions:
Updated session factories and types: Replaced
DbSessionFactoryandDbSessionwithAsyncDbSessionFactoryandAsyncDbSessionacross various modules (src/aleph/api_entrypoint.py,src/aleph/chains/bsc.py,src/aleph/chains/ethereum.py,src/aleph/chains/nuls2.py,src/aleph/chains/connector.py,src/aleph/chains/indexer_reader.py,src/aleph/chains/chain_data_service.py). [1] [2] [3] [4] [5] [6] [7]Replaced synchronous session management: Updated methods to use
async withfor session handling instead ofwith, ensuring proper handling of asynchronous database operations. This change affects methods likeadd_pending_tx,upsert_chain_sync_status, and others (src/aleph/chains/chain_data_service.py,src/aleph/chains/ethereum.py,src/aleph/chains/indexer_reader.py). [1] [2] [3]Migration of Database Operations to Async:
Async database operations: Replaced synchronous calls to database methods (e.g.,
session.commit(),upsert_file) with their asynchronous counterparts (await session.commit(),await upsert_file). This ensures compatibility with async workflows (src/aleph/chains/chain_data_service.py,src/aleph/chains/ethereum.py,src/aleph/chains/indexer_reader.py). [1] [2] [3]Async helper methods: Updated helper methods like
get_last_height,count_pending_txs, andget_unconfirmed_messagesto useawaitfor asynchronous execution (src/aleph/chains/ethereum.py,src/aleph/chains/indexer_reader.py). [1] [2] [3]Consistency Across Chain Modules:
src/aleph/chains/bsc.py,src/aleph/chains/ethereum.py,src/aleph/chains/nuls2.py,src/aleph/chains/connector.py). This ensures uniformity in database interaction patterns. [1] [2] [3] [4]These updates pave the way for better scalability and performance in asynchronous environments, aligning the codebase with modern best practices for database interaction.
Notes
This PR is the first parts of message parallelization, but this parts could have side effects intense testing will be needed
For some reason on local test in some case (rare case) it's gave me an error about psycopg2 but it's shouldn't be used since we use async engine that use asyncpg, not sure if it only on testing parts or if could happen in prod