5
\$\begingroup\$

I have a class that takes in two strings via the __init___ method and then does a bunch of string manipulation to return numerous other strings via instance methods. Some instance methods rely on other instance methods. Anyways, I feel like this could be made more pythonic and OO, but I'm quite bad at more advanced OO topics, like using @property, @staticmethod, @classmethod, etc.

ACTIVE_DATASETS = active.datasets

class FileMetadata:
    def __init__(self, object_bucket: str, object_location: str):
        self.object_bucket = object_bucket
        self.object_location = object_location
        self.date = self._date()
        self.source_bucket = self._source_bucket()
        self.source_object_location = self._source_object_location()
        self.source_object_path_prefix = self._source_object_path_prefix()
        self.source_object_filename = self._source_object_filename()
        self.source_object_extension = self._source_object_extension()
        self.destination_bucket = self._destination_bucket()
        self.destination_path_prefix = self._destination_path_prefix()
        self.destination_object_location = self._destination_object_location()
        self.datasets = self._get_datasets()

    def _match_inbound_csv_to_dataset(
        self,
    ):
        datasets = []
        for dataset in ACTIVE_DATASETS:
            if re.match(rf"{dataset.file_match_regex}", self.source_object_filename):
                datasets.append(dataset.dataset_name)

        if len(datasets) == 0:
            raise MatchingDatasetNotFound(
                f"No matching dataset found for inbound file: {self.source_object_filename}"
            )

        if len(datasets) > 1:
            raise MultipleMatchingDatasetsFound(
                f"Multiple matching datasets found for inbound file: {self.source_object_filename}"
            )

        if len(datasets) == 1:
            print(
                f"Matching dataset {datasets[0]} found for inbound file: {self.source_object_filename}"
            )
            return datasets

    def _get_datasets(self):
        if self.source_object_extension == "xlsx":
            return None
        else:
            return self._match_inbound_csv_to_dataset()

    def _date(self) -> str:
        return datetime.today().strftime("%Y-%m-%d")

    def _source_bucket(self) -> str:
        return self.object_bucket

    def _source_object_location(self) -> str:
        return self.object_location

    def _source_object_path_prefix(self) -> str:
        "Used for filtering objects not dropped in the proper folder"
        split = self.object_location.split("/")[:-1]
        path = "/".join(split) + "/"
        return path

    def _source_object_filename(self) -> str:
        return self.object_location.split("/")[-1]

    def _source_object_extension(self) -> str:
        split = self.source_object_filename.split(".")
        if len(split) == 1:
            return None
        return self.source_object_filename.split(".")[-1]

    def _destination_bucket(self) -> str:
        return os.environ.get(
            "EMPASSION_INGEST_BUCKET_NAME", "dev-ingest"
        )

    def _destination_path_prefix(self) -> str:
        return f"{self.source_bucket}/{self.date}"

    def _destination_object_location(self) -> str:
        return f"{self.source_bucket}/{self.date}/{self.source_object_filename}"

it's called and used like so:

class Input(BaseModel):
    object_bucket: str
    object_location: str


class Output(BaseModel):
    source_bucket: str
    source_object_location: str
    source_object_path_prefix: str
    source_object_filename: str
    source_object_extension: str
    destination_bucket: str
    destination_path_prefix: str
    destination_object_location: str
    datasets: List[str]


@http
def main(request):
    resp = Input(**request.get_json())
    file_metadata = FileMetadata(resp.object_bucket, resp.object_location)

    output = Output(
        source_bucket=file_metadata.source_bucket,
        source_object_location=file_metadata.source_object_location,
        source_object_path_prefix=file_metadata.source_object_path_prefix,
        source_object_filename=file_metadata.source_object_filename,
        source_object_extension=file_metadata.source_object_extension,
        destination_bucket=file_metadata.destination_bucket,
        destination_path_prefix=file_metadata.destination_path_prefix,
        destination_object_location=file_metadata.destination_object_location,
        datasets=file_metadata.datasets,
    )

    return json.dumps(output.model_dump()), 200



\$\endgroup\$
4
  • 2
    \$\begingroup\$ This is missing import statements. \$\endgroup\$ Commented Jul 20, 2023 at 21:01
  • 1
    \$\begingroup\$ To help reviewers give you better answers, we need to know what the code is intended to achieve. Please add sufficient context to your question to describe the purpose of the code. We want to know why much more than how. The more you tell us about what your code is for, the easier it will be for reviewers to help you. Also, edit the title to simply summarise the task, rather than your concerns about the code. \$\endgroup\$ Commented Jul 21, 2023 at 7:08
  • 1
    \$\begingroup\$ Incorporating advice from an answer into the question violates the question-and-answer nature of this site. You could post improved code as a new question, as an answer, or as a link to an external site - as described in I improved my code based on the reviews. What next?. I have rolled back the edit, so the answers make sense again. \$\endgroup\$ Commented Jul 21, 2023 at 19:10
  • \$\begingroup\$ sorry @TobySpeight i was unaware of that. thanks for rolling it back. \$\endgroup\$
    – metersk
    Commented Jul 21, 2023 at 19:34

2 Answers 2

7
\$\begingroup\$

Getters

You define a lot of "private" getters like self.date = self._date() which returns datetime.today.... Let's use a few @propertys for that

from pathlib import Path


class FileMetadata:
    def __init__(self, object_bucket: str, object_location: str):
        self.object_bucket = object_bucket
        self.object_location = object_location
        self.source_bucket = object_bucket

        # this has some nice convenience features
        p = Path(self.object_location)

        # like p.parent, p.name, and p.suffix
        self.source_object_path_prefix = f"{p.parent}/"
        self.source_object_filename = p.name
        self.source_object_extension = p.suffix

        self.destination_bucket = os.environ.get(
            "EMPASSION_INGEST_BUCKET_NAME", "dev-ingest"
        )

        self.datasets = self._get_datasets()

    @property
    def date(self):
        # do you care if this script runs past midnight?
        if not getattr(self, '_date', None):
            self._date = datetime.today().strftime("%Y-%m-%d")
        return self._date

    @property
    def source_object_location(self):
        return self.object_location

    @property
    def destination_path_prefix(self):
        return f"{self.source_bucket}/{self.date}"

    @property
    def destination_object_location(self):
        return (f"{self.destination_path_prefix}/{self.source_object_filename}"
        )

    def _get_datasets(self):
        if self.source_object_extension == "xlsx":
            return None
        else:
            return self._match_inbound_csv_to_dataset()

Now we have way fewer methods. While we're here, I've utilized pathlib.Path to take advantage of some convenience functions.

Repeated regexes

It might be useful to cache repeated regexes here:

class FileMetadata:
    dataset_cache = {}
 
    def __init__(self, ...):
        ~snip~

    def _match_inbound_csv_to_dataset(
        self,
    ):
        datasets = []
        for dataset in ACTIVE_DATASETS:
            name, regex = dataset.name, dataset.file_match_regex

            # pre-compile the regex
            if name not in self.dataset_cache:
                r = re.compile(regex)
                self.dataset_cache[name] = r
            else:
                r = self.dataset_cache[name]

            # use the compiled regex to match
            if r.match(self.source_object_filename):
                datasets.append(name)

Testing if a list is empty

Don't test if len(lst) == 0, use if lst:

        if not datasets:
            raise MatchingDatasetNotFound(
                f"No matching dataset found for inbound file: {self.source_object_filename}"
            )

self.datasets

I don't like the naming here, since you require only one dataset to match. I might be inclined to call it self.dataset, but it looks like it needs to be a list so I'll leave it alone. Let's use some unpacking to check the length of the list:

    def _match_inbound_csv_to_dataset(self):
        ~snip~
        dataset, *others = datasets

        if others:
            raise MultipleMatchingDatasetsFound(
                f"Multiple matching datasets found for inbound file: {self.source_object_filename}"
            )

        print(
            f"Matching dataset {dataset} found for inbound file: {self.source_object_filename}"
        )
        return [dataset]

cached_property

If you are on python 3.8+ you have access to functools.cached_property which helps simplify the date property:

from functools import cached_property


class FileMetadata:
    ~snip~ 

    @cached_property
    def date(self):
        return datetime.today().strftime("%Y-%m-%d")

    
\$\endgroup\$
3
  • 1
    \$\begingroup\$ this is very helpful. thank you! \$\endgroup\$
    – metersk
    Commented Jul 20, 2023 at 21:51
  • \$\begingroup\$ For date, you can use a cached_property instead of implementing caching yourself \$\endgroup\$ Commented Jul 21, 2023 at 15:42
  • \$\begingroup\$ @Brian61354270 true, totally forgot about that one \$\endgroup\$
    – C.Nivs
    Commented Jul 21, 2023 at 18:06
10
\$\begingroup\$

Your class is a collection of non-mutated members and transformation methods. That's pretty good already! Nice job.

Strengthen the guarantee of immutability by using NamedTuple as a superclass and having a class method pseudoconstructor, something like

class FileMetadata(NamedTuple):
    object_bucket: str
    object_location: str
    date: datetime.date
    # ...

    @classmethod
    def from_location(
        cls,
        object_bucket: str,
        object_location: str,
    ) -> 'FileMetadata':
        return cls(
            object_bucket=object_bucket,
            object_location=object_location,
            date=cls._date(),
            # ...
        )

For your path manipulation, use pathlib.Path instead of bare strings.

Yes, use @property for simple transformation methods like source_object_filename. Here is a demonstration for _source_object_path_prefix:

@property
def _source_object_path_prefix(self) -> Path:
    """
    Used for filtering objects not dropped in the proper folder
    Given /a/b/c.foo as an object_location, returns directory /a/b
    """
    return self.object_location.parent
\$\endgroup\$
5
  • \$\begingroup\$ would you be able to update your answer on how you would create the @property methods? \$\endgroup\$
    – metersk
    Commented Jul 21, 2023 at 20:15
  • 1
    \$\begingroup\$ @metersk See edit. I've attempted to do a 2-in-1 demo with @property and pathlib. \$\endgroup\$
    – Reinderien
    Commented Jul 22, 2023 at 19:57
  • \$\begingroup\$ Hmm I might be a bit confused. Isn’t object_location a string? How can I call .parent on it? \$\endgroup\$
    – metersk
    Commented Jul 23, 2023 at 20:24
  • \$\begingroup\$ As in my review - don't store it as a string; store it as a pathlib.Path \$\endgroup\$
    – Reinderien
    Commented Jul 24, 2023 at 0:37
  • \$\begingroup\$ Okay, I understand now. Thanks! \$\endgroup\$
    – metersk
    Commented Jul 24, 2023 at 15:16

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.