Class Serializer

dvc/database.py:49–73

@dataclass
class Serializer:
    sql: "Union[str, Selectable]"
    con: "Union[str, Connectable]"
    chunksize: int = 10_000

    def to_csv(self, file: StrOrBytesPath, progress=noop):
        import pandas as pd

        idfs = pd.read_sql(self.sql, self.con, chunksize=self.chunksize)
        with atomic_file(file) as f:
            for i, df in enumerate(idfs):
                df.to_csv(f, header=i == 0, index=False)
                progress(len(df))

    def to_json(self, file: StrOrBytesPath, progress=noop):  # noqa: ARG002
        import pandas as pd

        df = pd.read_sql(self.sql, self.con)
        with atomic_file(file) as f:
            df.to_json(f, orient="records")

    def export(self, file: StrOrBytesPath, format: str = "csv", progress=noop):  # noqa: A002
        if format == "json":
            return self.to_json(file, progress=progress)
        return self.to_csv(file, progress=progress)


@dataclass