Exporters

Exporters save lists of Pydantic models to files or databases. All exporters implement the same interface:

exporter.export(data: list[BaseModel], destination: str) -> None

• data — list of Pydantic models (must not be empty)
• destination — file path or table name (for SQL)

Import from steam_fetcher.exporters:

from steam_fetcher.exporters import JsonExporter, CsvExporter, ParquetExporter, SqlExporter

---

JsonExporter

Exports data as a JSON array. No extra dependencies.

from steam_fetcher.exporters import JsonExporter

exporter = JsonExporter(indent=4, ensure_ascii=False)
exporter.export([game], "games.json")

Parameter	Type	Default	Description
indent	int	4	JSON indentation level
ensure_ascii	bool	False	Escape non-ASCII characters

Output example:

[
    {
        "app_id": 730,
        "name": "Counter-Strike 2",
        "type": "game",
        "is_free": true
    }
]

---

CsvExporter

Exports data as CSV. Nested fields (lists, dicts) are flattened to strings. No extra dependencies.

from steam_fetcher.exporters import CsvExporter

exporter = CsvExporter(delimiter=",", flatten=True)
exporter.export([game], "games.csv")

Parameter	Type	Default	Description
delimiter	str	","	CSV delimiter
flatten	bool	True	Flatten nested fields to strings

Flattening rules:

• Lists → pipe-separated: ["Action", "FPS"] → "Action|FPS"
• Dicts → string representation: {"id": 1} → "{'id': 1}"
• None → empty string

---

ParquetExporter

Exports data as Apache Parquet via polars.

pip install steam-fetcher[parquet]

from steam_fetcher.exporters import ParquetExporter

exporter = ParquetExporter(compression="zstd")
exporter.export([game], "games.parquet")

Parameter	Type	Default	Description
compression	ParquetCompression	"zstd"	Compression algorithm

Supported compression values: "lz4", "uncompressed", "snappy", "gzip", "brotli", "zstd".

---

SqlExporter

Exports data to a SQL database via SQLAlchemy. Creates the table automatically.

pip install steam-fetcher[sql]

from steam_fetcher.exporters import SqlExporter

exporter = SqlExporter(
    url="sqlite:///steam.db",
    if_exists="append",
    chunk_size=1000,
)
exporter.export([game], "games")  # destination = table name

Parameter	Type	Default	Description
url	str	required	SQLAlchemy connection URL
if_exists	str	"append"	"fail", "replace", or "append"
chunk_size	int	1000	Records per INSERT batch

Connection URL examples:

sqlite:///games.db
postgresql://user:pass@localhost/steam
mysql://user:pass@localhost/steam

Nested fields (lists, dicts) are stored as their string representations. Column types are inferred from the first record (int → Integer, float → Float, bool → Boolean, everything else → Text).

---

Error handling

All exporters raise:

• ValueError("Data is empty") — if an empty list is passed
• ExportError — if the export operation fails (wraps the original exception)

from steam_fetcher.exporters import ExportError

try:
    exporter.export(data, "output.json")
except ValueError:
    print("No data to export")
except ExportError as e:
    print(f"Export failed: {e}")