Here are some suggestions to improve the code:
Use constants for file paths and other configurations: Use a constant for the configuration file path instead of hardcoding it in the code. This makes it easy to change the path in the future if needed.
Use a context manager for opening files: When opening files, use a context manager to ensure that the file is closed properly after use.
Validate SQL command and other parameters in the Report object: Perform validation of the SQL command, types, and format fields in the Report object during object creation. If validation fails, raise an appropriate exception.
Use a library to generate CSV files: Instead of manually generating a CSV file, use a library such as the
csv
module in Python to generate the file.Use a library to generate Excel files: Instead of manually generating an Excel file, use a library such as
openpyxl
orxlsxwriter
to generate the file.Add error handling: Add error handling to handle exceptions and errors that may occur during file generation and sending.
Use
send_mail
to send email: Use Django's built-insend_mail
function to send email instead of a customNotificationService
.
Here is the improved code:
pythonimport json
import csv
import io
from django.core.mail import send_mail
from django.db import connection
CONFIG_FILE_PATH = "/mnt/data/config.json"
FORBIDDEN_SQL_KEYWORDS = ["DROP", "UPDATE", "INSERT", "ALTER"]
VALID_REPORT_TYPES = ["csv", "excel"]
class Report:
def __init__(self, **kwargs):
self.name = kwargs.get("name")
self.label = kwargs.get("label")
self.database = kwargs.get("database")
self.sql_command = kwargs.get("sql_command")
self.types = kwargs.get("types")
self.format = kwargs.get("format")
self._validate()
def _validate(self):
if not all(getattr(self, field) for field in ("name", "label", "database", "sql_command", "types", "format")):
raise ValueError("All fields in Report object are required.")
if any(keyword in self.sql_command.upper() for keyword in FORBIDDEN_SQL_KEYWORDS):
raise ValueError("Invalid SQL command in Report object.")
if any(report_type not in VALID_REPORT_TYPES for report_type in self.types):
raise ValueError("Invalid report type in Report object.")
class ReportCollection:
def __init__(self):
self.reports = []
with open(CONFIG_FILE_PATH) as config_file:
config = json.load(config_file)
for report in config:
report_object = Report(**report)
self.reports.append(report_object)
def get_report_config(self, name):
for report in self.reports:
if report.name == name:
return report
class ReportingService:
def generate_and_send_report(self, name, email):
report = ReportCollection().get_report_config(name)
with connection.cursor() as cursor:
cursor.execute(report.sql_command)
results = cursor.fetchall()
for report_type in report.types:
file = self.generate_file_from_sql_results(results, report_type)
self.send_to_user(file, email)
def generate_file_from_sql_results(self, results, report_type):
if report_type == "csv":
output = io.StringIO()
writer = csv.writer(output)
writer.writerows(results)
file = output.getvalue().encode("utf-8")
elif report_type == "excel":
# logic for excel with openpyxl or xlsxwriter library
...
return file
def send_to_user(self