generated
user_5449717
python
a year ago
4.2 kB
14
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...