Repository: SuperDuperDB/superduperdb
File: superduper/components/listener.py (Python)

Code before:
    :param flatten: Flatten the output into separate records if ``True``.
    """

    type_id: t.ClassVar[str] = 'listener'
    breaks: t.ClassVar[t.Sequence[str]] = ('model', 'key', 'select')

    key: ModelInputType
    model: Model
    predict_kwargs: t.Optional[t.Dict] = dc.field(default_factory=dict)
    select: t.Optional[Query] = None
    cdc_table: str = ''
    output_table: t.Optional[Table] = None
    flatten: bool = False

    def _get_metadata(self):
        """Get metadata of the component."""
        metadata = super()._get_metadata()
        return {**metadata, 'output_table': self.output_table}

    def __post_init__(self, db):
        if not self.cdc_table and self.select:
            self.cdc_table = self.select.table
        self._set_upstream()

        return super().__post_init__(db)

    def handle_update_or_same(self, other):
        super().handle_update_or_same(other)
        other.output_table = self.output_table

    def dict(
        self, metadata: bool = True, defaults: bool = True, refs: bool = False
    ) -> t.Dict:
        """Convert to dictionary."""
        out = super().dict(metadata=metadata, defaults=defaults, refs=refs)
        if not metadata:
            try:
                del out['output_table']
            except KeyError:
                logging.warn('output_table not found in listener.dict()')
                pass
        return out

    def _set_upstream(self):
        deps = self.dependencies
        if deps:
            if not self.upstream:
                self.upstream = []
            try:
                it = 0
                for dep in deps:

Review comment (suggestion):
Avoid using `tuple` during creation to prevent diffs.

The issue occurs because tuples are converted to lists after saving, causing diffs.

Code after:
    :param flatten: Flatten the output into separate records if ``True``.
    """

    type_id: t.ClassVar[str] = 'listener'
    breaks: t.ClassVar[t.Sequence[str]] = ('model', 'key', 'select')

    key: ModelInputType
    model: Model
    predict_kwargs: t.Optional[t.Dict] = dc.field(default_factory=dict)
    select: t.Optional[Query] = None
    cdc_table: str = ''
    output_table: t.Optional[Table] = None
    flatten: bool = False

    def _get_metadata(self):
        """Get metadata of the component."""
        metadata = super()._get_metadata()
        return {**metadata, 'output_table': self.output_table}

    def __post_init__(self, db):
        if not self.cdc_table and self.select:
            self.cdc_table = self.select.table
        self._set_upstream()
        if isinstance(self.key, tuple):
            self.key = list(self.key)

        return super().__post_init__(db)

    def handle_update_or_same(self, other):
        super().handle_update_or_same(other)
        other.output_table = self.output_table

    def dict(
        self, metadata: bool = True, defaults: bool = True, refs: bool = False
    ) -> t.Dict:
        """Convert to dictionary."""
        out = super().dict(metadata=metadata, defaults=defaults, refs=refs)
        if not metadata:
            try:
                del out['output_table']
            except KeyError:
                logging.warn('output_table not found in listener.dict()')
                pass
        return out

    def _set_upstream(self):
        deps = self.dependencies
        if deps:
            if not self.upstream:
                self.upstream = []
            try: