(
self,
sql: "Union[str, Selectable]",
file: StrOrBytesPath,
format: str = "csv", # noqa: A002
progress=noop,
)
| 92 | raise DvcException("The database returned the following error") from exc |
| 93 | |
| 94 | def export( |
| 95 | self, |
| 96 | sql: "Union[str, Selectable]", |
| 97 | file: StrOrBytesPath, |
| 98 | format: str = "csv", # noqa: A002 |
| 99 | progress=noop, |
| 100 | ) -> None: |
| 101 | con = self.engine.connect() |
| 102 | if format == "csv": |
| 103 | con = con.execution_options(stream_results=True) # use server-side cursors |
| 104 | |
| 105 | with con: |
| 106 | serializer = Serializer(sql, con) |
| 107 | return serializer.export(file, format=format, progress=progress) |
| 108 | |
| 109 | |
| 110 | @contextmanager |
no test coverage detected