"""
MySQL Tools Module
Provides MySQL CRUD operations for subscription management
"""
from src.mysql_db import execute_query
from datetime import datetime, timedelta
def create_subscription(data):
"""
Create a new subscription record
Args:
data: Dictionary with subscription details
"""
try:
query = """
INSERT INTO subscriptions
(service, amount, currency, renewal_date, plan_name, cycle, status, email_id, email_date, category, notes)
VALUES (%(service)s, %(amount)s, %(currency)s, %(renewal_date)s, %(plan_name)s, %(cycle)s, %(status)s, %(email_id)s, %(email_date)s, %(category)s, %(notes)s)
"""
params = {
'service': data.get('service'),
'amount': data.get('amount'),
'currency': data.get('currency', 'USD'),
'renewal_date': data.get('renewal_date'),
'plan_name': data.get('plan_name', ''),
'cycle': data.get('cycle', 'monthly'),
'status': data.get('status', 'active'),
'email_id': data.get('email_id', ''),
'email_date': data.get('email_date'),
'category': data.get('category', ''),
'notes': data.get('notes', '')
}
result = execute_query(query, params)
return {
'success': True,
'subscription_id': result['last_insert_id'],
'message': f"Subscription created for {data.get('service')}"
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
def get_subscription(subscription_id):
"""
Get subscription by ID
Args:
subscription_id: Subscription ID
"""
try:
query = "SELECT * FROM subscriptions WHERE id = %s"
result = execute_query(query, (subscription_id,), fetch_one=True)
if not result:
return {
'success': False,
'error': f"Subscription {subscription_id} not found"
}
return {
'success': True,
'data': result
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
def update_subscription(subscription_id, data):
"""
Update existing subscription
Args:
subscription_id: Subscription ID
data: Dictionary with fields to update
"""
try:
set_clauses = []
params = []
allowed_fields = ['service', 'amount', 'currency', 'renewal_date', 'plan_name', 'cycle', 'status', 'category', 'notes']
for field in allowed_fields:
if field in data:
set_clauses.append(f"{field} = %s")
params.append(data[field])
if not set_clauses:
return {
'success': False,
'error': 'No valid fields to update'
}
params.append(subscription_id)
query = f"UPDATE subscriptions SET {', '.join(set_clauses)} WHERE id = %s"
result = execute_query(query, params)
return {
'success': True,
'affected_rows': result['affected_rows'],
'message': f"Subscription {subscription_id} updated"
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
def delete_subscription(subscription_id):
"""
Delete subscription
Args:
subscription_id: Subscription ID
"""
try:
query = "DELETE FROM subscriptions WHERE id = %s"
result = execute_query(query, (subscription_id,))
return {
'success': True,
'affected_rows': result['affected_rows'],
'message': f"Subscription {subscription_id} deleted"
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
def list_all_subscriptions(status='active', limit=100):
"""
List all subscriptions with optional filtering
Args:
status: Filter by status (default: active)
limit: Maximum number of results
"""
try:
query = """
SELECT * FROM subscriptions
WHERE status = %s
ORDER BY renewal_date ASC
LIMIT %s
"""
results = execute_query(query, (status, limit), fetch_all=True)
return {
'success': True,
'subscriptions': results,
'count': len(results)
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
def get_upcoming_renewals(days=7):
"""
Get subscriptions renewing within specified days
Args:
days: Number of days to look ahead
"""
try:
query = """
SELECT * FROM subscriptions
WHERE renewal_date BETWEEN CURDATE() AND DATE_ADD(CURDATE(), INTERVAL %s DAY)
AND status = 'active'
ORDER BY renewal_date ASC
"""
results = execute_query(query, (days,), fetch_all=True)
total_cost = sum(sub['amount'] for sub in results)
return {
'success': True,
'subscriptions': results,
'count': len(results),
'total_cost': float(total_cost)
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
def get_subscription_by_service(service_name):
"""
Get subscriptions by service name
Args:
service_name: Name of the service
"""
try:
query = "SELECT * FROM subscriptions WHERE service LIKE %s"
results = execute_query(query, (f"%{service_name}%",), fetch_all=True)
return {
'success': True,
'subscriptions': results,
'count': len(results)
}
except Exception as error:
return {
'success': False,
'error': str(error)
}
