Exporters
The tables extension provides several built-in exporters to export the table data in different formats. You can specify the exporters to be used in the exporters attribute of the TableDefinition.
Each exporter is a class that inherits from BaseExporter and implements the export method.
To use an exporter, you need to import the desired exporter from ckanext.tables.shared and add it to the exporters list in the TableDefinition.
import ckanext.tables.shared as t
class MyTable(t.TableDefinition):
def __init__(self):
super().__init__(
...
exporters=[
t.exporters.CSVExporter,
t.exporters.TSVExporter,
]
)
CSV, TSV, and NDJSON stream their output row by row. The other exporters (JSON, XLSX, YAML, HTML, PDF) build the full result set in memory before returning it, so exporting a very large table in one of those formats uses proportionally more memory.
Below you can see the source code for the base exporter class.
ExporterBase
Base class for table data exporters.
Source code in ckanext/tables/exporters.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85 | class ExporterBase:
"""Base class for table data exporters."""
name: str
label: str
mime_type: str
@classmethod
def is_available(cls) -> bool:
"""Whether this exporter's dependencies are installed.
Checked by the view *before* building the response — export() runs inside a
streamed generator body by the time it would raise, which is too late to
change the response's status code (headers are already committed by then).
Override for an exporter with an optional dependency.
"""
return True
@classmethod
def export(cls, table: "TableDefinition", params: "QueryParams") -> bytes:
"""Export the table data.
Args:
table: The table definition.
params: The query parameters.
Returns:
The exported data as bytes.
"""
raise NotImplementedError
@classmethod
def export_stream(cls, table: "TableDefinition", params: "QueryParams") -> Iterator[bytes]:
"""Yield the exported data in chunks.
The default just wraps ``export()`` as a single chunk; override this (and, for
the reverse direction, have ``export()`` join it) for exporters that can be
produced incrementally, so the response can start sending before the whole
file is built and without holding two full copies of it in memory at once.
"""
yield cls.export(table, params)
@classmethod
def get_table_columns(cls, table: "TableDefinition") -> list["ColumnDefinition"]:
"""Get the list of table columns to be exported.
Returns:
A list of column field names.
"""
# avoid circular import
from ckanext.tables.table import COLUMN_ACTIONS_FIELD # noqa PLC0415
return [col for col in table.columns if col.field != COLUMN_ACTIONS_FIELD]
|
List of Built-in Exporters
Below is a list of the available built-in exporters along with a brief description of each.
CSV Exporter
CSV exporter for table data.
Source code in ckanext/tables/exporters.py
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107 | class CSVExporter(ExporterBase):
"""CSV exporter for table data."""
name = "csv"
label = tk._("CSV")
mime_type = "text/csv"
@classmethod
def export(cls, table: "TableDefinition", params: "QueryParams") -> bytes:
return b"".join(cls.export_stream(table, params))
@classmethod
def export_stream(cls, table: "TableDefinition", params: "QueryParams") -> Iterator[bytes]:
writer = csv.writer(_Echo())
columns = cls.get_table_columns(table)
yield writer.writerow([col.title for col in columns]).encode("utf-8")
for row in table.get_raw_data(params, paginate=False):
yield writer.writerow([row.get(col.field, "") for col in columns]).encode("utf-8")
|
TSV Exporter
TSV exporter for table data.
Source code in ckanext/tables/exporters.py
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179 | class TSVExporter(ExporterBase):
"""TSV exporter for table data."""
name = "tsv"
label = tk._("TSV")
mime_type = "text/tab-separated-values"
@classmethod
def export(cls, table: "TableDefinition", params: "QueryParams") -> bytes:
return b"".join(cls.export_stream(table, params))
@classmethod
def export_stream(cls, table: "TableDefinition", params: "QueryParams") -> Iterator[bytes]:
writer = csv.writer(_Echo(), delimiter="\t")
columns = cls.get_table_columns(table)
yield writer.writerow([col.title for col in columns]).encode("utf-8")
for row in table.get_raw_data(params, paginate=False):
yield writer.writerow([row.get(col.field, "") for col in columns]).encode("utf-8")
|
JSON Exporter
JSON exporter for table data.
Source code in ckanext/tables/exporters.py
110
111
112
113
114
115
116
117
118
119
120
121 | class JSONExporter(ExporterBase):
"""JSON exporter for table data."""
name = "json"
label = tk._("JSON")
mime_type = "application/json"
@classmethod
def export(cls, table: "TableDefinition", params: "QueryParams") -> bytes:
data = table.get_raw_data(params, paginate=False)
return json.dumps(data, default=str).encode("utf-8")
|
XLSX Exporter
Excel (XLSX) exporter for table data.
Source code in ckanext/tables/exporters.py
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157 | class XLSXExporter(ExporterBase):
"""Excel (XLSX) exporter for table data."""
name = "xlsx"
label = tk._("Excel")
mime_type = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
@classmethod
def is_available(cls) -> bool:
return importlib.util.find_spec("openpyxl") is not None
@classmethod
def export(cls, table: "TableDefinition", params: "QueryParams") -> bytes:
if not cls.is_available():
raise ImportError("openpyxl is required for XLSX export but is not installed.")
from openpyxl import Workbook # noqa: PLC0415
wb = Workbook()
ws = wb.active
ws.title = "Data" # type: ignore
columns = cls.get_table_columns(table)
header = [col.title for col in columns]
ws.append(header) # type: ignore
# Write data rows
data = table.get_raw_data(params, paginate=False)
for row in data:
ws.append([row.get(col.field, "") for col in columns]) # type: ignore
output = BytesIO()
wb.save(output)
return output.getvalue()
|
HTML Exporter
HTML exporter for table data.
Source code in ckanext/tables/exporters.py
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235 | class HTMLExporter(ExporterBase):
"""HTML exporter for table data."""
name = "html"
label = tk._("HTML")
mime_type = "text/html"
@classmethod
def export(cls, table: "TableDefinition", params: "QueryParams") -> bytes:
data = table.get_raw_data(params, paginate=False)
columns = cls.get_table_columns(table)
headers = [col.title for col in columns]
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
return tk.render(
"tables/exporters/html_export.html",
{
"data": data,
"headers": headers,
"table": table,
"timestamp": timestamp,
"columns": columns,
},
).encode("utf-8")
|
PDF Exporter
Requires the optional weasyprint dependency (pip install ckanext-tables[pdf]).
PDF exporter for table data.
Source code in ckanext/tables/exporters.py
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258 | class PDFExporter(ExporterBase):
"""PDF exporter for table data."""
name = "pdf"
label = tk._("PDF")
mime_type = "application/pdf"
@classmethod
def is_available(cls) -> bool:
return importlib.util.find_spec("weasyprint") is not None
@classmethod
def export(cls, table: "TableDefinition", params: "QueryParams") -> bytes:
if not cls.is_available():
raise ImportError("WeasyPrint is required for PDF export but is not installed.")
from weasyprint import HTML # noqa: PLC0415
# reuse HTML exporter template for PDF generation
html_content = HTMLExporter.export(table, params).decode("utf-8")
return HTML(string=html_content).write_pdf() or b""
|
YAML Exporter
YAML exporter for table data.
Source code in ckanext/tables/exporters.py
182
183
184
185
186
187
188
189
190
191
192 | class YAMLExporter(ExporterBase):
"""YAML exporter for table data."""
name = "yaml"
label = tk._("YAML")
mime_type = "application/x-yaml"
@classmethod
def export(cls, table: "TableDefinition", params: "QueryParams") -> bytes:
data = table.get_raw_data(params, paginate=False)
return yaml.safe_dump(data, allow_unicode=True).encode("utf-8")
|
NDJSON Exporter
NDJSON exporter for table data.
Source code in ckanext/tables/exporters.py
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209 | class NDJSONExporter(ExporterBase):
"""NDJSON exporter for table data."""
name = "ndjson"
label = tk._("NDJSON")
mime_type = "application/x-ndjson"
@classmethod
def export(cls, table: "TableDefinition", params: "QueryParams") -> bytes:
return b"".join(cls.export_stream(table, params))
@classmethod
def export_stream(cls, table: "TableDefinition", params: "QueryParams") -> Iterator[bytes]:
for row in table.get_raw_data(params, paginate=False):
yield (json.dumps(row, default=str) + "\n").encode("utf-8")
|