feat: celery example improvements (#14)

* feat: celery transaction minor x02

* feat: celery transaction minor x03

Author: Niennienzz

celery-transaction-processing/README.md

@@ -3,7 +3,7 @@ pip install -r requirements.txt
 ```
 
 ```bash
-uvicorn main:app --reload
+uvicorn main:api_app --reload
 ```
 
 ```bash

celery-transaction-processing/main.py

@@ -11,20 +11,24 @@
 from deps import get_deps, get_constants
 from tasks import reconcile_transaction
 
-app = FastAPI()
+api_app = FastAPI()
 
 
-@app.post("/transactions")
-async def transaction(
+@api_app.post("/transactions")
+async def create_transaction(
         req: utils.TransactionRequest,
         db: Session = Depends(get_deps().get_db_session),
         df: Dragonfly = Depends(get_deps().get_dragonfly),
         w3: Web3 = Depends(get_deps().get_web3),
 ) -> utils.TransactionResponse:
     # Try to acquire a lock on the user account.
     lock_key = utils.user_account_lock_key(req.user_account_id)
-    lock = df.set(name=lock_key, value=utils.LOCK_VALUE, nx=True, ex=utils.LOCK_EXPIRATION_SECONDS)
-    if not lock:
+    locked = df.set(
+        name=lock_key, value=utils.LOCK_VALUE,
+        nx=True, ex=utils.LOCK_EXPIRATION_SECONDS,
+    )
+
+    if not locked:
         raise HTTPException(
             status_code=409,
             detail="User account is locked since a transaction is submitted very recently. Please try again later.",
@@ -81,8 +85,8 @@ async def transaction(
 
     # Cache the transaction in Dragonfly.
     cache_key = utils.txn_cache_key(txn.id)
-    _ = df.hset(cache_key, mapping=utils.txn_to_dict(txn))
-    _ = df.expire(cache_key, utils.CACHE_NORMAL_EXPIRATION_SECONDS)
+    mapping = utils.txn_to_dict(txn)
+    utils.hset_and_expire(df, cache_key, mapping, utils.CACHE_NORMAL_EXPIRATION_SECONDS)
 
     # Start the transaction reconciliation task.
     reconcile_transaction.delay(txn.id)
@@ -91,7 +95,7 @@ async def transaction(
     return utils.txn_to_response(txn)
 
 
-@app.get("/transactions/{txn_id}")
+@api_app.get("/transactions/{txn_id}")
 async def get_transaction(
         txn_id: int,
         db: Session = Depends(get_deps().get_db_session),
@@ -115,11 +119,10 @@ async def get_transaction(
     # If the transaction is not found, cache an empty value with only the ID and return a 404.
     # Caching an empty value is important to prevent cache penetrations.
     if txn is None:
-        _ = df.hset(cache_key, mapping={"id": txn_id})
-        _ = df.expire(cache_key, utils.CACHE_EMPTY_EXPIRATION_SECONDS)
+        utils.hset_and_expire(df, cache_key, {"id": txn_id}, utils.CACHE_EMPTY_EXPIRATION_SECONDS)
         raise HTTPException(status_code=404, detail="Transaction not found")
 
     # Cache the transaction in Dragonfly and return the response.
-    _ = df.hset(cache_key, mapping=utils.txn_to_dict(txn))
-    _ = df.expire(cache_key, utils.CACHE_NORMAL_EXPIRATION_SECONDS)
+    mapping = utils.txn_to_dict(txn)
+    utils.hset_and_expire(df, cache_key, mapping, utils.CACHE_NORMAL_EXPIRATION_SECONDS)
     return utils.txn_to_response(txn)

celery-transaction-processing/models.sql

@@ -8,14 +8,16 @@ CREATE TABLE user_accounts
 CREATE TABLE user_account_transactions
 (
     id                                INTEGER PRIMARY KEY,
-    user_account_id                   INTEGER REFERENCES user_accounts (id) NOT NULL,
-    transaction_hash                  TEXT                                  NOT NULL,
-    from_public_address               TEXT                                  NOT NULL,
-    to_public_address                 TEXT                                  NOT NULL,
-    transaction_amount_in_wei         INTEGER                               NOT NULL,
-    transaction_fee_total_in_wei      INTEGER                               NOT NULL,
-    transaction_fee_blockchain_in_wei INTEGER                               NOT NULL,
-    status                            TEXT                                  NOT NULL DEFAULT 'PENDING' CHECK (status IN ('PENDING', 'SUCCESSFUL', 'FAILED'))
+    user_account_id                   INTEGER NOT NULL,
+    transaction_hash                  TEXT    NOT NULL,
+    from_public_address               TEXT    NOT NULL,
+    to_public_address                 TEXT    NOT NULL,
+    transaction_amount_in_wei         INTEGER NOT NULL,
+    transaction_fee_total_in_wei      INTEGER NOT NULL,
+    transaction_fee_blockchain_in_wei INTEGER NOT NULL,
+    status                            TEXT    NOT NULL DEFAULT 'PENDING' CHECK (status IN ('PENDING', 'SUCCESSFUL', 'FAILED')),
+
+    FOREIGN KEY (user_account_id) REFERENCES user_accounts (id)
 );
 
 CREATE INDEX idx_user_account_id ON user_account_transactions (user_account_id);

celery-transaction-processing/tasks.py

@@ -23,7 +23,8 @@ def __init__(self, message):
         super().__init__(message)
 
 
-# Use Dragonfly as the broker and the result backend for Celery.
+# Celery app configuration.
+# Use Dragonfly as the message broker and the result storage backend for Celery.
 # Dragonfly is wire-protocol compatible with Redis.
 # We can use the same Redis URL(s) as long as Dragonfly is running on the port specified.
 app = Celery(
@@ -32,22 +33,28 @@ def __init__(self, message):
     backend=get_constants().get_celery_backend_url(),
 )
 
-TASK_MAX_RETRIES: Final[int] = 20
-TASK_RETRY_DELAY: Final[int] = 100
+TASK_MAX_RETRIES: Final[int] = 32
+TASK_RETRY_BACKOFF: Final[int] = 100
+TASK_RETRY_BACKOFF_MAX: Final[int] = 1600
 
 
 # Define a Celery task to reconcile a transaction.
 # We check the transaction status from the blockchain and update the database accordingly.
 # The blockchain is just an example. Many finical systems have similar reconciliation or settlement processes.
 #
-# Maximum retry count is TASK_MAX_RETRIES, and the delay between retries is exponential.
-# The retry delays are [1, 2, 4, 8, 16, 32, 64, 128, 256, 512, ...] * TASK_RETRY_DELAY seconds.
-# Thus, the first retry is after 100 seconds, the second retry is after 200 seconds, and so on.
+# Maximum retry count is 32, and the delay between retries is exponential.
+#
+# The retry delays are [1, 2, 4, 8, 16, ...] x 100 seconds.
+# So, the first retry is after 100 seconds, the second retry is after 200 seconds, and so on.
+#
+# The maximum backoff delay is set to 1600 seconds.
+# Thus, the retry delays in this configuration are [100, 200, 400, 800, 1600, 1600, ...] seconds up to 32 retries.
 @app.task(
     bind=True,
     autoretry_for=(TaskRetryException,),
     max_retries=TASK_MAX_RETRIES,
-    retry_backoff=TASK_RETRY_DELAY,
+    retry_backoff=TASK_RETRY_BACKOFF,
+    retry_backoff_max=TASK_RETRY_BACKOFF_MAX,
     retry_jitter=False,
 )
 def reconcile_transaction(_self, txn_id: str):

celery-transaction-processing/utils.py

@@ -2,6 +2,8 @@
 from typing import Final
 from urllib.parse import urlparse
 
+from redis import Redis as Dragonfly
+
 import models
 
 CACHE_NORMAL_EXPIRATION_SECONDS: Final[int] = 60
@@ -112,3 +114,15 @@ def parse_dragonfly_url(dragonfly_url) -> DragonflyURLParsed:
         port = 6379
 
     return DragonflyURLParsed(host=host, port=port, db=db)
+
+
+def hset_and_expire(
+        df: Dragonfly,
+        key: str,
+        mapping: dict,
+        expiration: int,
+):
+    pipe = df.pipeline()
+    pipe.hset(key, mapping=mapping)
+    pipe.expire(key, expiration)
+    pipe.execute()