# Imports
from django.conf import settings
import os
from superdjango.utils import read_csv, read_file, File
# Exports
__all__ = (
"CSVDatabase",
"CSVError",
)
# Classes
[docs]class CSVDatabase(File):
"""The database for CSV error capture."""
[docs] def __init__(self, path):
"""Initialize the database.
:param path: The path to the CSV file.
:type path: str
"""
self.errors = list()
self.is_loaded = False
super().__init__(path)
def __iter__(self):
return iter(self.errors)
def __len__(self):
return len(self.errors)
[docs] def fetch(self, identifier):
"""Get the error instance.
:param identifier: The unique identifier of the error.
:type identifier: str
:rtype: CSVError | None
"""
for e in self.errors:
if e.identifier == identifier:
return e
return None
[docs] def load(self):
"""Load the database.
:rtype: bool
"""
if not self.exists:
return False
rows = read_csv(self.path, first_row_field_names=True)
for row in rows:
path = os.path.join(settings.SUPERDJANGO_ERROR_CAPTURE_OUTPUT_PATH, "tracebacks", "%s.txt" % row['id'])
if not os.path.exists(path):
continue
traceback = read_file(path)
error = CSVError(
added_dt=row['date'],
identifier=row['id'],
name=row['error'],
path=row['path'],
line=row['line'],
traceback=traceback
)
self.errors.append(error)
self.is_loaded = len(self.errors) > 0
return self.is_loaded
[docs]class CSVError(object):
"""An individual error captured and stored in the CSV database."""
[docs] def __init__(self, **kwargs):
self._attributes = kwargs
def __getattr__(self, item):
return self._attributes.get(item)
@property
def file_name(self):
"""Get the file name where the error occurred.
:rtype: str
"""
return self.path.split("/")[-1]