fix: lint fixes

Author: amalcaraz

deployment/migrations/versions/0033_1c06d0ade60c_calculate_costs_statically.py

@@ -7,7 +7,6 @@
 """
 
 from decimal import Decimal
-from typing import Dict
 from alembic import op
 import sqlalchemy as sa
 import logging
@@ -18,7 +17,7 @@
 from aleph.db.accessors.messages import get_message_by_item_hash
 from aleph.services.cost import _is_confidential_vm, get_detailed_costs, CostComputableContent
 from aleph.services.pricing_utils import build_default_pricing_model
-from aleph.types.cost import ProductPriceType, ProductPricing
+from aleph.types.cost import ProductPriceType
 from aleph.types.db_session import DbSession
 
 logger = logging.getLogger("alembic")

src/aleph/services/pricing_utils.py

@@ -5,7 +5,10 @@
 import datetime as dt
 from typing import Dict, List, Union
 
-from aleph.db.accessors.aggregates import get_aggregate_elements, merge_aggregate_elements
+from aleph.db.accessors.aggregates import (
+    get_aggregate_elements,
+    merge_aggregate_elements,
+)
 from aleph.db.models import AggregateElementDb
 from aleph.toolkit.constants import (
     DEFAULT_PRICE_AGGREGATE,
@@ -16,22 +19,24 @@
 from aleph.types.db_session import DbSession
 
 
-def build_pricing_model_from_aggregate(aggregate_content: Dict[Union[ProductPriceType, str], dict]) -> Dict[ProductPriceType, ProductPricing]:
+def build_pricing_model_from_aggregate(
+    aggregate_content: Dict[Union[ProductPriceType, str], dict]
+) -> Dict[ProductPriceType, ProductPricing]:
     """
     Build a complete pricing model from an aggregate content dictionary.
-    
-    This function converts the DEFAULT_PRICE_AGGREGATE format or any pricing aggregate 
-    content into a dictionary of ProductPricing objects that can be used by the cost 
+
+    This function converts the DEFAULT_PRICE_AGGREGATE format or any pricing aggregate
+    content into a dictionary of ProductPricing objects that can be used by the cost
     calculation functions.
-    
+
     Args:
         aggregate_content: Dictionary containing pricing information with ProductPriceType as keys
-    
+
     Returns:
         Dictionary mapping ProductPriceType to ProductPricing objects
     """
     pricing_model: Dict[ProductPriceType, ProductPricing] = {}
-    
+
     for price_type, pricing_data in aggregate_content.items():
         try:
             price_type = ProductPriceType(price_type)
@@ -41,16 +46,17 @@ def build_pricing_model_from_aggregate(aggregate_content: Dict[Union[ProductPric
         except (KeyError, ValueError) as e:
             # Log the error but continue processing other price types
             import logging
+
             logger = logging.getLogger(__name__)
             logger.warning(f"Failed to parse pricing for {price_type}: {e}")
-    
+
     return pricing_model
 
 
 def build_default_pricing_model() -> Dict[ProductPriceType, ProductPricing]:
     """
     Build the default pricing model from DEFAULT_PRICE_AGGREGATE constant.
-    
+
     Returns:
         Dictionary mapping ProductPriceType to ProductPricing objects
     """
@@ -60,55 +66,57 @@ def build_default_pricing_model() -> Dict[ProductPriceType, ProductPricing]:
 def get_pricing_aggregate_history(session: DbSession) -> List[AggregateElementDb]:
     """
     Get all pricing aggregate updates in chronological order.
-    
+
     Args:
         session: Database session
-        
+
     Returns:
         List of AggregateElementDb objects ordered by creation_datetime
     """
     aggregate_elements = get_aggregate_elements(
-        session=session,
-        owner=PRICE_AGGREGATE_OWNER,
-        key=PRICE_AGGREGATE_KEY
+        session=session, owner=PRICE_AGGREGATE_OWNER, key=PRICE_AGGREGATE_KEY
     )
     return list(aggregate_elements)
 
 
-def get_pricing_timeline(session: DbSession) -> List[tuple[dt.datetime, Dict[ProductPriceType, ProductPricing]]]:
+def get_pricing_timeline(
+    session: DbSession,
+) -> List[tuple[dt.datetime, Dict[ProductPriceType, ProductPricing]]]:
     """
     Get the complete pricing timeline with timestamps and pricing models.
-    
+
     This function returns a chronologically ordered list of pricing changes,
-    useful for processing messages in chronological order and applying the 
+    useful for processing messages in chronological order and applying the
     correct pricing at each point in time.
-    
+
     This properly merges aggregate elements up to each point in time to create
     the cumulative pricing state, similar to how _update_aggregate works.
-    
+
     Args:
         session: Database session
-        
+
     Returns:
         List of tuples containing (timestamp, pricing_model)
     """
     pricing_elements = get_pricing_aggregate_history(session)
-    
+
     timeline = []
-    
+
     # Add default pricing as the initial state
-    timeline.append((dt.datetime.min.replace(tzinfo=dt.timezone.utc), build_default_pricing_model()))
-    
+    timeline.append(
+        (dt.datetime.min.replace(tzinfo=dt.timezone.utc), build_default_pricing_model())
+    )
+
     # Build cumulative pricing models by merging elements up to each timestamp
     elements_so_far = []
     for element in pricing_elements:
         elements_so_far.append(element)
-        
+
         # Merge all elements up to this point to get the cumulative state
         merged_content = merge_aggregate_elements(elements_so_far)
-        
+
         # Build pricing model from the merged content
         pricing_model = build_pricing_model_from_aggregate(merged_content)
         timeline.append((element.creation_datetime, pricing_model))
-    
-    return timeline
+
+    return timeline

src/aleph/toolkit/constants.py

@@ -1,16 +1,18 @@
+from typing import Dict, Union
+
+from aleph.types.cost import ProductPriceType
+
 KiB = 1024
 MiB = 1024 * 1024
 GiB = 1024 * 1024 * 1024
 
 MINUTE = 60
 HOUR = 60 * MINUTE
 
-from aleph.types.cost import ProductPriceType
-
 PRICE_AGGREGATE_OWNER = "0xFba561a84A537fCaa567bb7A2257e7142701ae2A"
 PRICE_AGGREGATE_KEY = "pricing"
 PRICE_PRECISION = 18
-DEFAULT_PRICE_AGGREGATE = {
+DEFAULT_PRICE_AGGREGATE: Dict[Union[ProductPriceType, str], dict] = {
     ProductPriceType.PROGRAM: {
         "price": {
             "storage": {"payg": "0.000000977", "holding": "0.05"},
@@ -50,7 +52,9 @@
             "memory_mib": 2048,
         },
     },
-    ProductPriceType.WEB3_HOSTING: {"price": {"fixed": 50, "storage": {"holding": "0.333333333"}}},
+    ProductPriceType.WEB3_HOSTING: {
+        "price": {"fixed": 50, "storage": {"holding": "0.333333333"}}
+    },
     ProductPriceType.PROGRAM_PERSISTENT: {
         "price": {
             "storage": {"payg": "0.000000977", "holding": "0.05"},

src/aleph/web/controllers/prices.py

@@ -21,16 +21,17 @@
 )
 from aleph.services.cost import (
     _get_product_price_type,
-    _get_settings,
     get_detailed_costs,
     get_payment_type,
     get_total_and_detailed_costs,
     get_total_and_detailed_costs_from_db,
 )
 from aleph.services.pricing_utils import get_pricing_timeline
+from aleph.toolkit.constants import DEFAULT_SETTINGS_AGGREGATE
 from aleph.toolkit.costs import format_cost_str
 from aleph.types.db_session import DbSession
 from aleph.types.message_status import MessageStatus
+from aleph.types.settings import Settings
 from aleph.web.controllers.app_state_getters import (
     get_session_factory_from_request,
     get_storage_service_from_request,
@@ -174,7 +175,7 @@ async def message_price_estimate(request: web.Request):
 
 async def recalculate_message_costs(request: web.Request):
     """Force recalculation of message costs in chronological order with historical pricing.
-    
+
     This endpoint will:
     1. Get all messages that need cost recalculation (if item_hash provided, just that message)
     2. Get the pricing timeline to track price changes over time
@@ -183,15 +184,15 @@ async def recalculate_message_costs(request: web.Request):
     5. Delete existing cost entries and recalculate with historical pricing
     6. Store the new cost calculations
     """
-    
+
     session_factory = get_session_factory_from_request(request)
-    
+
     # Check if a specific message hash was provided
     item_hash_param = request.match_info.get("item_hash")
-    
+
     with session_factory() as session:
         messages_to_recalculate: List[MessageDb] = []
-        
+
         if item_hash_param:
             # Recalculate costs for a specific message
             try:
@@ -203,78 +204,99 @@ async def recalculate_message_costs(request: web.Request):
             # Recalculate costs for all executable messages, ordered by time (oldest first)
             select_stmt = (
                 select(MessageDb)
-                .where(MessageDb.type.in_([MessageType.instance, MessageType.program, MessageType.store]))
+                .where(
+                    MessageDb.type.in_(
+                        [MessageType.instance, MessageType.program, MessageType.store]
+                    )
+                )
                 .order_by(MessageDb.time.asc())
             )
             result = session.execute(select_stmt)
             messages_to_recalculate = result.scalars().all()
-        
+
         if not messages_to_recalculate:
             return web.json_response(
-                {"message": "No messages found for cost recalculation", "recalculated_count": 0}
+                {
+                    "message": "No messages found for cost recalculation",
+                    "recalculated_count": 0,
+                }
             )
-        
+
         # Get the pricing timeline to track price changes over time
         pricing_timeline = get_pricing_timeline(session)
         LOGGER.info(f"Found {len(pricing_timeline)} pricing changes in timeline")
-        
+
         recalculated_count = 0
         errors = []
         current_pricing_model = None
         current_pricing_index = 0
-        
+
         for message in messages_to_recalculate:
             try:
                 # Find the applicable pricing model for this message's timestamp
-                while (current_pricing_index < len(pricing_timeline) - 1 and 
-                       pricing_timeline[current_pricing_index + 1][0] <= message.time):
+                while (
+                    current_pricing_index < len(pricing_timeline) - 1
+                    and pricing_timeline[current_pricing_index + 1][0] <= message.time
+                ):
                     current_pricing_index += 1
-                
+
                 current_pricing_model = pricing_timeline[current_pricing_index][1]
                 pricing_timestamp = pricing_timeline[current_pricing_index][0]
-                
-                LOGGER.debug(f"Message {message.item_hash} at {message.time} using pricing from {pricing_timestamp}")
-                
+
+                LOGGER.debug(
+                    f"Message {message.item_hash} at {message.time} using pricing from {pricing_timestamp}"
+                )
+
                 # Delete existing cost entries for this message
                 delete_costs_for_message(session, message.item_hash)
-                
+
                 # Get the message content and determine product type
                 content: ExecutableContent = message.parsed_content
-                product_type = _get_product_price_type(content, None, current_pricing_model)
-                
+
+                # TODO: Calculate settings timeline
+                settings = Settings.from_aggregate(DEFAULT_SETTINGS_AGGREGATE)
+
+                product_type = _get_product_price_type(
+                    content, settings, current_pricing_model
+                )
+
                 # Get the pricing for this specific product type
                 if product_type not in current_pricing_model:
-                    LOGGER.warning(f"Product type {product_type} not found in pricing model for message {message.item_hash}")
+                    LOGGER.warning(
+                        f"Product type {product_type} not found in pricing model for message {message.item_hash}"
+                    )
                     continue
-                
+
                 pricing = current_pricing_model[product_type]
-                
+
                 # Calculate new costs using the historical pricing model
-                new_costs = get_detailed_costs(session, content, message.item_hash, pricing)
-                
+                new_costs = get_detailed_costs(
+                    session, content, message.item_hash, pricing
+                )
+
                 if new_costs:
                     # Store the new cost calculations
                     upsert_stmt = make_costs_upsert_query(new_costs)
                     session.execute(upsert_stmt)
-                
+
                 recalculated_count += 1
-                
+
             except Exception as e:
                 error_msg = f"Failed to recalculate costs for message {message.item_hash}: {str(e)}"
                 LOGGER.error(error_msg)
                 errors.append({"item_hash": message.item_hash, "error": str(e)})
-        
+
         # Commit all changes
         session.commit()
-        
+
         response_data = {
             "message": "Cost recalculation completed with historical pricing",
             "recalculated_count": recalculated_count,
             "total_messages": len(messages_to_recalculate),
-            "pricing_changes_found": len(pricing_timeline)
+            "pricing_changes_found": len(pricing_timeline),
         }
-        
+
         if errors:
             response_data["errors"] = errors
-        
+
         return web.json_response(response_data)