generated
user_5449717
python
6 months ago
4.2 kB
6
Indexable
testing
Generated Action Class: ```python from datetime import datetime from typing import Dict, Any import re from action_handler import register_action class SkipDaysAction: def __init__(self, core_service: Any, electricity_service: Any): """ Initialize the SkipDaysAction class. Args: - core_service (Any): An instance of the CoreService class. - electricity_service (Any): An instance of the ElectricityService class. """ self.core_service = core_service self.electricity_service = electricity_service @register_action(actions=["add_skip_data", "delete_skip_data"]) async def handle(self, context: Dict[str, Any]) -> Dict[str, str]: """ Handle the action from the context. Args: - context (Dict[str, Any]): A dictionary containing the action and parameters. Returns: - Dict[str, str]: A dictionary containing the status and message of the action. """ try: action = context["actions"][0]["action"] parameters = context["actions"][0]["parameters"] # Validate parameters await self.validate_parameters(parameters) # Process the action if action == "add_skip_data": # Implement logic for adding skip data # For demonstration purposes, we'll just return a success message return {"status": "success", "message": "Skip data added successfully"} elif action == "delete_skip_data": # Implement logic for deleting skip data # For demonstration purposes, we'll just return a success message return {"status": "success", "message": "Skip data deleted successfully"} else: return {"status": "error", "message": "Invalid action"} except ValueError as ve: return {"status": "error", "message": str(ve)} except Exception as e: return {"status": "error", "message": "An error occurred while processing the action"} async def validate_parameters(self, parameters: Dict[str, str]) -> None: """ Validate the parameters. Args: - parameters (Dict[str, str]): A dictionary containing the parameters. Raises: - ValueError: If any of the parameters are invalid. """ required_fields = ["company_id", "task_id", "start_date", "end_date"] # Check for required fields for field in required_fields: if field not in parameters: raise ValueError(f"Missing required field: {field}") # Validate dates try: start_date = datetime.strptime(parameters["start_date"], "%Y-%m-%d") end_date = datetime.strptime(parameters["end_date"], "%Y-%m-%d %H:%M") except ValueError: raise ValueError("Invalid date format. Please use YYYY-MM-DD for start_date and YYYY-MM-DD HH:MM for end_date") # Check if dates are in the past if start_date > datetime.now() or end_date > datetime.now(): raise ValueError("Dates must be in the past") # Check if end_date includes 23:00 timestamp if end_date.hour != 23 or end_date.minute != 0: raise ValueError("end_date must include 23:00 timestamp") # Check if date format is correct if not re.match(r"\d{4}-\d{2}-\d{2}", parameters["start_date"]): raise ValueError("Invalid date format for start_date. Please use YYYY-MM-DD") if not re.match(r"\d{4}-\d{2}-\d{2} 23:00", parameters["end_date"]): raise ValueError("Invalid date format for end_date. Please use YYYY-MM-DD 23:00") ``` Example usage: ```python skip_days_action = SkipDaysAction(core_service, electricity_service) context = { "actions": [ { "action": "add_skip_data", "parameters": { "company_id": "company123", "task_id": "task456", "start_date": "2024-03-01", "end_date": "2024-03-10 23:00" } } ] } result = skip_days_action.handle(context) print(result) ```
Editor is loading...