[Backend] Delete unused, reformate

web_app/api/main.py

@@ -48,9 +48,6 @@
     },
 )
 
-# Set up the templates directory
-BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-
 # Add session middleware with a secret key
 app.add_middleware(SessionMiddleware, secret_key=f"Secret:{str(uuid4())}")
 # CORS middleware for React frontend

web_app/api/serializers/transaction.py

@@ -97,19 +97,3 @@ class UpdateUserContractRequest(BaseModel):
 
     wallet_id: str
     contract_address: str
-
-
-class DeploymentStatus(BaseModel):
-    """
-    Pydantic model for the deployment status.
-    """
-
-    is_contract_deployed: bool
-
-
-class ContractAddress(BaseModel):
-    """
-    Pydantic model for the contract address.
-    """
-
-    contract_address: str | None

web_app/api/telegram.py

@@ -98,7 +98,7 @@ async def telegram_webhook(update: Update):
         return b"", 200
 
 
-@router.post(
+@router.post( # FIXME REMOVE IT (delete and frontend, not used)
     "/api/telegram/save-user",
     tags=["Telegram Operations"],
     summary="Save or update Telegram user information",

web_app/api/user.py

@@ -159,12 +159,21 @@ async def subscribe_to_notification(
     Success status of the subscription.
     """
     user = user_db.get_user_by_wallet_id(data.wallet_id)
+    # Check if the user exists; if not, raise a 404 error
     if not user:
         raise HTTPException(status_code=404, detail="User not found")
-    is_allowed_notification = telegram_db.allow_notification(data.telegram_id)
 
-    if is_allowed_notification:
+    telegram_id = data.telegram_id
+    # Is not provided, attempt to retrieve it from the database
+    if not telegram_id:
+        tg_user = telegram_db.get_telegram_user_by_wallet_id(data.wallet_id)
+        telegram_id = tg_user.telegram_id
+    # Is found, set the notification preference for the user
+    if telegram_id:
+        telegram_db.set_allow_notification(telegram_id, data.wallet_id)
         return {"detail": "User subscribed to notifications successfully"}
+
+    # If no Telegram ID is available, raise
     raise HTTPException(
         status_code=400, detail="Failed to subscribe user to notifications"
     )
@@ -246,56 +255,3 @@ async def get_stats() -> GetStatsResponse:
     except Exception as e:
         logger.error(f"Error in get_stats: {e}")
         raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
-
-
-@router.get(
-    "/api/get-user-history",
-    tags=["User Operations"],
-    summary="Get user position history",
-    response_model=UserHistoryResponse,
-    response_description="List of user positions including status,created_at, \
-                                start_price, amount, and multiplier.",
-)
-async def get_user_history(user_id: str) -> list[dict]:
-    """
-    Retrieves the history of positions for a specified user.
-
-    ### Parameters:
-    - **user_id**: The unique ID of the user whose position history is being fetched.
-
-    ### Returns:
-    - A list of positions with the following details:
-        - `status`: Current status of the position.
-        - `created_at`: Timestamp when the position was created.
-        - `start_price`: Initial price of the asset when the position was opened.
-        - `amount`: Amount involved in the position.
-        - `multiplier`: Leverage multiplier applied to the position.
-    """
-    # FIXME REMOVE IT
-    try:
-        # Fetch user history from the database
-        positions = user_db.fetch_user_history(user_id)
-
-        if not positions:
-            logger.info(f"No positions found for user_id={user_id}")
-            return []
-
-        return positions
-
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
-
-
-@router.post("/allow-notification/{telegram_id}")
-async def allow_notification(
-    telegram_id: int,
-    telegram_db: TelegramUserDBConnector = Depends(lambda: TelegramUserDBConnector()),
-):
-    """Enable notifications for a specific telegram user"""
-    try:
-        telegram_db.allow_notification(telegram_id=telegram_id)
-        return {"message": "Notifications enabled successfully"}
-    except ValueError as e:
-        raise HTTPException(status_code=404, detail=str(e))
-    except Exception as e:
-        raise HTTPException(status_code=500, detail="Internal server error")

web_app/contract_tools/mixins/dashboard.py

@@ -83,15 +83,6 @@ async def get_wallet_balances(cls, holder_address: str) -> Dict[str, str]:
 
         return wallet_balances
 
-    @classmethod
-    def _get_products(cls, dapps: list) -> list[dict]:
-        """
-        Get the products from the dapps.
-        :param dapps: List of dapps
-        :return: List of positions
-        """
-        return [product for dapp in dapps for product in dapp.get("products", [])]
-
     @classmethod
     def _calculate_sum(
         cls, price: Decimal, amount: Decimal, multiplier: Decimal

web_app/db/crud/airdrop.py

@@ -62,9 +62,9 @@ def delete_all_users_airdrop(self, user_id: uuid.UUID) -> None:
         """
         with self.Session() as db:
             try:
-                airdrops = db.query(AirDrop).filter_by(user_id=user_id).all()
-                for airdrop in airdrops:
-                    db.delete(airdrop)
+                db.query(AirDrop).filter_by(user_id=user_id).delete(
+                    synchronize_session=False
+                )
                 db.commit()
             except SQLAlchemyError as e:
                 logger.error(f"Error deleting airdrops for user {user_id}: {str(e)}")

web_app/db/crud/telegram.py

@@ -20,6 +20,14 @@ class TelegramUserDBConnector(DBConnector):
     Provides database connection and operations management for the TelegramUser model.
     """
 
+    def get_telegram_user_by_wallet_id(self, wallet_id: str) -> TelegramUser | None:
+        """
+        Retrieves a TelegramUser by their wallet ID.
+        :param wallet_id: str
+        :return: TelegramUser | None
+        """
+        return self.get_object_by_field(TelegramUser, "wallet_id", wallet_id)
+
     def get_user_by_telegram_id(self, telegram_id: str) -> TelegramUser | None:
         """
         Retrieves a TelegramUser by their Telegram ID.
@@ -91,46 +99,11 @@ def set_allow_notification(self, telegram_id: str, wallet_id: str) -> bool:
         """
         Set wallet_id and is_allowed_notification to True for a user by their telegram ID.
         """
-        with self.Session() as session:
-            if telegram_user := self.get_user_by_telegram_id(telegram_id):
-                telegram_user.is_allowed_notification = True
-                session.commit()
-                logger.info(f"Notification allowed for user with telegram_id {telegram_id}")
-                return telegram_user
-            else:
-                logger.info(f"User with telegram_id {telegram_id} not found, creating new one")
-                self.create_telegram_user(
-                    dict(telegram_id=telegram_id, wallet_id=wallet_id)
-                )
-                self.allow_notification(telegram_id)
-
-    def allow_notification(self, telegram_id: str) -> bool:
-        """
-        Update is_allowed_notification field to True for a specific telegram user
-
-        Args:
-            telegram_id: Telegram user ID
-
-        Raises:
-            ValueError: If the user with the given telegram_id is not found
-        """
-        with self.Session() as session:
-            user = (
-                session.query(TelegramUser).filter_by(telegram_id=telegram_id).first()
+        self.save_or_update_user(
+            dict(
+                telegram_id=telegram_id,
+                wallet_id=wallet_id,
+                is_allowed_notification=True,
             )
-            if not user:
-                raise ValueError(f"User with telegram_id {telegram_id} not found")
-
-            user.is_allowed_notification = True
-            session.commit()
-            return True
-
-    def is_allowed_notification(self, wallet_id: str = None) -> bool | None:
-        """
-        Returns true or false if a telegram user allowed notification.
-
-        Args:
-            wallet_id: Wallet ID of the user.
-        """
-        user = self.get_object_by_field(TelegramUser, "wallet_id", wallet_id)
-        return user.is_allowed_notification if user else None
+        )
+        return True

web_app/db/crud/user.py

@@ -19,27 +19,6 @@ class UserDBConnector(DBConnector):
     Provides database connection and operations management for the User model.
     """
 
-    def get_all_users_with_opened_position(self) -> List[User]:
-        """
-        Retrieves all users with an OPENED position status from the database.
-        First queries Position table for OPENED positions, then gets the associated users.
-
-        :return: List[User]
-        """
-        with self.Session() as db:
-            try:
-                users = (
-                    db.query(User)
-                    .join(Position, Position.user_id == User.id)
-                    .filter(Position.status == Status.OPENED.value)
-                    .distinct()
-                    .all()
-                )
-                return users
-            except SQLAlchemyError as e:
-                logger.error(f"Error retrieving users with OPENED positions: {e}")
-                return []
-
     def get_users_for_notifications(self) -> List[Tuple[str, str]]:
         """
         Retrieves the contract_address of users with an OPENED position status and
@@ -118,31 +97,6 @@ def get_unique_users_count(self) -> int:
                 logger.error(f"Failed to retrieve unique users count: {str(e)}")
                 return 0
 
-    def delete_user_by_wallet_id(self, wallet_id: str) -> None:
-        """
-        Deletes a user from the database by their wallet ID.
-        Rolls back the transaction if the operation fails.
-
-        :param wallet_id: str
-        :return: None
-        :raises SQLAlchemyError: If the operation fails
-        """
-        with self.Session() as session:
-            try:
-                user = session.query(User).filter(User.wallet_id == wallet_id).first()
-                if user:
-                    session.delete(user)
-                    session.commit()
-                    logger.info(
-                        f"User with wallet_id {wallet_id} deleted successfully."
-                    )
-                else:
-                    logger.warning(f"No user found with wallet_id {wallet_id}.")
-            except SQLAlchemyError as e:
-                session.rollback()
-                logger.error(f"Failed to delete user with wallet_id {wallet_id}: {e}")
-                raise e
-
     def fetch_user_history(self, user_id: int) -> List[dict]:
         """
         Fetches all positions for a user with the specified fields: