Skip to content

feat: Prevent duplicated rules to be created #20

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
danoctua merged 1 commit into from
Apr 17, 2025
Merged

feat: Prevent duplicated rules to be created #20

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
227 changes: 224 additions & 3 deletions backend/core/actions/chat/rule/blockchain.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,13 +100,70 @@ def _resolve_collection_address(

return None

def check_duplicate(
self,
chat_id: int,
address_raw: str | None,
asset: NftCollectionAsset | None,
category: NftCollectionCategoryType | None,
entity_id: int | None = None,
) -> None:
"""
Checks for duplicate rules in the Telegram chat collection service. This function ensures
that no duplicate rules of the same type and category exist for a specific chat. If such
a rule exists, an exception is raised instructing the user to modify the existing rule
instead of creating a new one.

This method performs the check by retrieving existing rules based on the provided chat ID,
address, asset, and category, filtering out rules with the same ID as the supplied entity ID
(if provided), and checking for duplicates.

:param chat_id: The unique identifier for the Telegram chat where the rule applies.
:param address_raw: The address associated with the rule (optional).
:param asset: The asset associated with the rule, of type NftCollectionAsset (optional).
:param category: The category of the currency to which the rule applies, defined
as NftCollectionCategoryType (optional).
:param entity_id: The unique identifier for an existing rule to exclude from duplicate
checks (optional).
:return: None. The function raises an exception if a duplicate rule exists.
:raises HTTPException: Raised if a rule of the same type and category already exists
for the specified chat, excluding the rule with the provided `entity_id`.
"""
existing_rules = self.telegram_chat_nft_collection_service.find(
chat_id=chat_id,
address=address_raw,
asset=asset,
category=category,
)
if next(filter(lambda rule: rule.id != entity_id, existing_rules), None):
raise HTTPException(
detail="Rule of that type and category already exists for that chat. Please, modify it instead.",
status_code=HTTP_400_BAD_REQUEST,
)

async def create(
self,
asset: NftCollectionAsset | None,
address_raw: str | None,
category: NftCollectionCategoryType | None,
threshold: int,
) -> NftEligibilityRuleDTO:
"""
Creates a new NFT eligibility rule by linking a chat to an NFT collection. It resolves
the NFT collection's address based on the provided parameters and ensures that there
are no duplicate rules for the same chat and collection.

:param asset: Represents an optional NFT collection asset. This asset is used to
resolve the NFT collection address.
:param address_raw: The raw address of the NFT collection. This may be used in resolving
the final collection address.
:param category: Represents an optional category of the NFT collection.
:param threshold: An integer value specifying the threshold condition for the rule.

:return: A data transfer object (DTO) representing the created NFT eligibility rule.
The DTO encapsulates all properties of the rule created for the chat and
NFT collection.
"""
address = self._resolve_collection_address(address_raw, asset, category)

if not address:
Expand All @@ -118,7 +175,13 @@ async def create(
status_code=HTTP_400_BAD_REQUEST,
)

await self.nft_collection_action.get_or_create(address)
nft_collection_dto = await self.nft_collection_action.get_or_create(address)
self.check_duplicate(
chat_id=self.chat.id,
address_raw=nft_collection_dto.address,
asset=asset,
category=category,
)

new_rule = self.telegram_chat_nft_collection_service.create(
CreateTelegramChatNFTCollectionRuleDTO(
Expand All @@ -144,6 +207,30 @@ async def update(
threshold: int,
is_enabled: bool,
) -> NftEligibilityRuleDTO:
"""
Updates an existing NFT eligibility rule for a specific chat.

This method retrieves the rule by ID and updates it with the provided
details. It ensures the NFT collection address is resolved based on
the provided asset and category, creates or retrieves the corresponding
NFT collection if necessary, and checks for duplicate rules. The updated
rule is then returned as a data transfer object (DTO).

:param rule_id: The unique identifier of the rule to be updated.
:param asset: The NFT collection asset associated with the rule,
or None if not provided.
:param address_raw: The raw address of the NFT collection, or None.
:param category: The NFT collection category type associated with the
rule, or None if not provided.
:param threshold: The minimum threshold value for rule activation.
:param is_enabled: A boolean flag indicating whether the rule is
currently enabled.
:return: A data transfer object (DTO) representing the updated
NFT eligibility rule.
:raises HTTPException: If the rule with the provided ID does not exist,
there is a problem resolving the NFT collection address or
rule duplication is detected.
"""
try:
rule = self.telegram_chat_nft_collection_service.get(
rule_id, chat_id=self.chat.id
Expand All @@ -165,7 +252,13 @@ async def update(
status_code=HTTP_400_BAD_REQUEST,
)

await self.nft_collection_action.get_or_create(address)
nft_collection_dto = await self.nft_collection_action.get_or_create(address)
self.check_duplicate(
chat_id=self.chat.id,
address_raw=nft_collection_dto.address,
asset=asset,
category=category,
)

rule = self.telegram_chat_nft_collection_service.update(
rule=rule,
Expand Down Expand Up @@ -205,13 +298,63 @@ def read(self, rule_id: int) -> ChatEligibilityRuleDTO:
)
return ChatEligibilityRuleDTO.from_jetton_rule(rule)

def check_duplicate(
self,
chat_id: int,
address_raw: str,
category: CurrencyCategory | None,
entity_id: int | None = None,
) -> None:
"""
Checks for duplicate rules in the system based on provided criteria.

This method verifies whether a duplicate rule exists for the given chat ID,
address, and category, excluding the rule specified by the entity ID. If a
duplicate is found, it raises an HTTPException indicating the conflict.

This is typically used to enforce uniqueness constraints and ensure that no
redundant rules are added to the system.

:param chat_id: ID of the Telegram chat for which the rule is being verified
:param address_raw: Address of the entity to be checked for duplicate rules
:param category: Category of the currency to filter the rules
:param entity_id: Identifier of the existing rule to exclude from duplicate
checks; this can be None if no existing rule is to be excluded
:return: None. The function raises an error if a duplicate rule is found.
"""
existing_rules = self.telegram_chat_jetton_service.find(
chat_id=chat_id,
address=address_raw,
category=category,
)
if next(filter(lambda rule: rule.id != entity_id, existing_rules), None):
raise HTTPException(
detail="Rule of that type and category already exists for that chat. Please, modify it instead.",
status_code=HTTP_400_BAD_REQUEST,
)

async def create(
self,
address_raw: str,
category: CurrencyCategory | None,
threshold: float | int,
) -> ChatEligibilityRuleDTO:
"""
Creates and associates a new chat-eligibility rule for a specific jetton, with
options to set a category and a threshold. Ensures duplication prevention
before rule creation. Logs the operation's activity and returns the created
rule mapped to a DTO.

:param address_raw: The raw address of the jetton to associate.
:param category: The category of the currency or jetton, if applicable.
:param threshold: The minimum threshold value to set for the rule.
:return: A data transfer object (DTO) representing the created chat-eligibility
rule linked to the jetton.
:raises HTTPException: If there is a problem resolving the jetton address or
the rule duplication is detected.
"""
jetton_dto = await self.jetton_action.get_or_create(address_raw)
self.check_duplicate(self.chat.id, jetton_dto.address, category)

new_rule = self.telegram_chat_jetton_service.create(
CreateTelegramChatJettonRuleDTO(
Expand All @@ -233,16 +376,37 @@ async def update(
threshold: int | float,
is_enabled: bool,
) -> ChatEligibilityRuleDTO:
"""
Updates an existing chat jetton rule with specified parameters.

This method fetches the existing rule using the provided `rule_id` and updates it with
new values supplied as arguments. It also ensures no duplicate rules exist with the
same parameters. If the rule does not exist, an HTTPException with status code 404 is
raised. The updated rule information is wrapped into a `ChatEligibilityRuleDTO` and
returned.

:param rule_id: Identifier of the rule to be updated
:param address_raw: Raw address to be associated with the rule
:param category: Category of the currency, could be optional
:param threshold: Threshold value for the rule, could be an integer or float
:param is_enabled: Boolean flag indicating if the rule is enabled
:return: A `ChatEligibilityRuleDTO` object representing the updated rule
:raises HTTPException: If the rule with the provided `rule_id` does not exist,
if there is a problem resolving the jetton address or rule duplication is detected.
"""
try:
rule = self.telegram_chat_jetton_service.get(rule_id, chat_id=self.chat.id)
except NoResultFound:
raise HTTPException(
detail="Rule not found",
status_code=HTTP_404_NOT_FOUND,
)

jetton_dto = await self.jetton_action.get_or_create(address_raw)

self.check_duplicate(
self.chat.id, jetton_dto.address, category, entity_id=rule.id
)

updated_rule = self.telegram_chat_jetton_service.update(
rule=rule,
dto=UpdateTelegramChatJettonRuleDTO(
Expand Down Expand Up @@ -279,11 +443,54 @@ def read(self, rule_id: int) -> ChatEligibilityRuleDTO:
)
return ChatEligibilityRuleDTO.from_toncoin_rule(rule)

def check_duplicate(
self,
chat_id: int,
category: CurrencyCategory | None,
entity_id: int | None = None,
) -> None:
"""
Checks for existing rules in the service to determine if a duplicate rule exists for the specified
chat, category, and optionally the entity ID. If a duplicate is found, an HTTPException is raised.

:param chat_id: The identifier of the chat to check for duplicate rules.
:param category: The category of the rule to check for duplication.
:param entity_id: The unique identifier of the rule entity. It is optional and defaults to None.
:return: None. It raises an HTTPException if a duplicate rule is found.
:raises HTTPException: If a duplicate rule of the specified type and category is found.
"""
existing_rules = self.telegram_chat_toncoin_service.find(
chat_id=chat_id,
category=category,
)
if next(filter(lambda rule: rule.id != entity_id, existing_rules), None):
raise HTTPException(
detail="Rule of that type and category already exists for that chat. Please, modify it instead.",
status_code=HTTP_400_BAD_REQUEST,
)

def create(
self,
category: CurrencyCategory | None,
threshold: float | int,
) -> ChatEligibilityRuleDTO:
"""
Creates a new chat eligibility rule based on the specified category and
threshold.

This method verifies if there is an existing rule for the specified category
in the current chat. If no duplicate exists, it creates and associates a
new TON rule with the chat while enabling the rule by default. The method
logs the creation event and returns a data transfer object (DTO) that
represents the newly created TON rule.

:param category: The currency category to associate with the new TON rule.
:param threshold: The minimum threshold value for the TON rule.
:return: A DTO representing the chat eligibility rule created from the
TON rule.
:raises HTTPException: If a duplicate rule of the specified type and category is found.
"""
self.check_duplicate(chat_id=self.chat.id, category=category)
new_rule = self.telegram_chat_toncoin_service.create(
CreateTelegramChatToncoinRuleDTO(
chat_id=self.chat.id,
Expand All @@ -302,13 +509,27 @@ def update(
threshold: int | float,
is_enabled: bool,
) -> ChatEligibilityRuleDTO:
"""
Updates an existing chat eligibility rule for TON based on the provided attributes.
Retrieves the rule by its ID and ensures it belongs to the current chat. Checks for
duplicate rules before applying updates.

:param rule_id: The unique identifier of the rule to be updated.
:param category: The category of the currency for the rule, or None if not applicable.
:param threshold: The threshold value associated with the rule, could be an integer or float.
:param is_enabled: A boolean indicating whether the rule is enabled or disabled.
:return: A data transfer object (DTO) representing the updated chat eligibility rule for TON.
:raises HTTPException: If the rule with the provided ID does not exist,
or if a duplicate rule of the specified type and category is found.
"""
try:
rule = self.telegram_chat_toncoin_service.get(rule_id, chat_id=self.chat.id)
except NoResultFound:
raise HTTPException(
detail="Rule not found",
status_code=HTTP_404_NOT_FOUND,
)
self.check_duplicate(chat_id=self.chat.id, category=category, entity_id=rule.id)

updated_rule = self.telegram_chat_toncoin_service.update(
rule=rule,
Expand Down
26 changes: 25 additions & 1 deletion backend/core/services/chat/rule/blockchain.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import logging
from abc import ABC
from typing import Any

from sqlalchemy import desc

Expand All @@ -23,7 +24,9 @@
logger = logging.getLogger(__name__)


TelegramChatRuleType = TelegramChatJetton | TelegramChatNFTCollection
TelegramChatRuleType = (
TelegramChatJetton | TelegramChatNFTCollection | TelegramChatToncoin
)
CreateTelegramChatRuleDTOType = (
CreateTelegramChatJettonRuleDTO
| CreateTelegramChatNFTCollectionRuleDTO
Expand Down Expand Up @@ -53,6 +56,27 @@ def get(self, id_: int, chat_id: int) -> TelegramChatRuleType:
.one()
)

def find(self, **params: Any) -> list[type[TelegramChatRuleType]]:
"""
Find records in the database table based on filter parameters provided as a dictionary.

This method allows querying the database using parameters corresponding to column names in
the model's table. Invalid parameters (not matching any column in the table) will raise an
exception. The method filters records based on these valid parameters and returns a list
of results.

:param params: kwargs containing column names as keys and filter values as values.
Only valid column names of the model table are allowed.
:return: A list of records filtered by the provided parameters.
:raises AttributeError: If any of the specified keys in `params` does not match any column
name of the model's table.
"""
for param in params.keys():
if param not in self.model.__table__.columns.keys():
raise AttributeError(f"Invalid parameter {param!r}.")

return self.db_session.query(self.model).filter_by(**params).all()

def update(
self,
rule: TelegramChatRuleType,
Expand Down
Loading