When it is necessary to select aggregates from a repository based on a condition that matches a computed field of the aggregate, some code “duplication” occurs.

For example:

class Element:
  status: Status
  updated_at: Time

  def is_ready(self) -> bool:
    return self.status == Ready and self.updated_at < now().sub(1 * time.Hour)

It is necessary to implement a method in the repository to find “ready” Elements:

class ElementRepository:
  def get_ready(self) -> List[Element]:

To do this, an SQL query must be implemented with duplicated conditions WHERE status = 'ready' AND updated_at < (NOW() - INTERVAL '1 hour'). If the logic of get_ready changes, the SQL query will also need to be changed.

Is there a way to describe this logic in one place?