APIリファレンス

メイン設定クラス

EstatDltConfig

e-Stat APIからdltへのデータ統合のメイン設定クラスです。ソースとロード先の設定を組み合わせ、データ抽出とロードのための追加処理オプションを提供します。

Bases: BaseModel

Main configuration for e-Stat API to DLT integration.

Combines source and destination configurations with additional processing options for data extraction and loading.

Attributes:

Name	Type	Description
`source`	`SourceConfig`	e-Stat API source configuration.
`destination`	`DestinationConfig`	DLT destination configuration.
`batch_size`	`Optional[int]`	Number of records per batch.
`max_retries`	`int`	Maximum API retry attempts.
`timeout`	`Optional[int]`	API request timeout in seconds.

Source code in src/estat_api_dlt_helper/config/models.py

class EstatDltConfig(BaseModel):
    """Main configuration for e-Stat API to DLT integration.

    Combines source and destination configurations with additional
    processing options for data extraction and loading.

    Attributes:
        source: e-Stat API source configuration.
        destination: DLT destination configuration.
        batch_size: Number of records per batch.
        max_retries: Maximum API retry attempts.
        timeout: API request timeout in seconds.
    """

    source: SourceConfig = Field(..., description="e-Stat API source configuration")
    destination: DestinationConfig = Field(
        ..., description="DLT destination configuration"
    )

    # Optional processing configuration
    batch_size: Optional[int] = Field(
        default=None, description="Number of records to process in each batch"
    )
    max_retries: int = Field(default=3, description="Maximum number of API retry attempts")
    timeout: Optional[int] = Field(default=None, description="API request timeout in seconds")

    # Data transformation options
    flatten_metadata: bool = Field(
        default=False, description="Whether to flatten metadata into table columns"
    )
    include_api_metadata: bool = Field(
        default=True, description="Whether to include API response metadata in the table"
    )

    model_config = ConfigDict(
        validate_assignment=True,
        extra="forbid",
    )

SourceConfig

e-Stat APIデータソースの設定クラスです。認証、取得する統計表選択、各種オプションを含む統計データの取得パラメータを定義します。

パラメータの詳細はe_Stat API 仕様を参照のこと。

Bases: BaseModel

Configuration for e-Stat API data source.

Defines parameters for fetching statistical data from e-Stat API, including authentication, data selection, and pagination options.

Attributes:

Name	Type	Description
`app_id`	`str`	e-Stat API application ID for authentication.
`statsDataId`	`Union[str, List[str]]`	Statistical table ID(s) to fetch.
`lang`	`Literal['J', 'E']`	Language for API response (J: Japanese, E: English).
`metaGetFlg`	`Literal['Y', 'N']`	Whether to fetch metadata.
`cntGetFlg`	`Literal['Y', 'N']`	Whether to fetch only record count.

Source code in src/estat_api_dlt_helper/config/models.py

class SourceConfig(BaseModel):
    """Configuration for e-Stat API data source.

    Defines parameters for fetching statistical data from e-Stat API,
    including authentication, data selection, and pagination options.

    Attributes:
        app_id: e-Stat API application ID for authentication.
        statsDataId: Statistical table ID(s) to fetch.
        lang: Language for API response (J: Japanese, E: English).
        metaGetFlg: Whether to fetch metadata.
        cntGetFlg: Whether to fetch only record count.
    """

    app_id: str = Field(..., description="e-Stat API application ID (API key)")
    statsDataId: Union[str, List[str]] = Field(
        ...,
        description="Statistical table ID(s) to fetch. Can be a single ID or list of IDs",
    )
    lang: Literal["J", "E"] = Field(default="J", description="Language of the API response")
    metaGetFlg: Literal["Y", "N"] = Field(
        default="Y", description="Whether to fetch metadata (Y/N)"
    )
    cntGetFlg: Literal["Y", "N"] = Field(
        default="N", description="Whether to fetch only count (Y/N)"
    )

    # 以下オプションパラメータ
    explanationGetFlg: Optional[Literal["Y", "N"]] = Field(
        default="Y", description="Whether to fetch explanations (Y/N)"
    )
    annotationGetFlg: Optional[Literal["Y", "N"]] = Field(
        default="Y", description="Whether to fetch annotations (Y/N)"
    )
    replaceSpChars: Literal["0", "1", "2", "3"] = Field(
        default="2",
        description="Special character replacement mode | 0: 置換しない, 1: 置換する, 2: NULL, 3: 'NA'",
    )

    # データ選択パラメータ
    lvTab: Optional[str] = Field(default=None, description="Table level")
    cdTab: Optional[str] = Field(default=None, description="Table code")
    cdTabFrom: Optional[str] = Field(default=None, description="Table code from")
    cdTabTo: Optional[str] = Field(default=None, description="Table code to")
    lvTime: Optional[str] = Field(default=None, description="Time level")
    cdTime: Optional[str] = Field(default=None, description="Time code")
    cdTimeFrom: Optional[str] = Field(default=None, description="Time code from")
    cdTimeTo: Optional[str] = Field(default=None, description="Time code to")
    lvArea: Optional[str] = Field(default=None, description="Area level")
    cdArea: Optional[str] = Field(default=None, description="Area code")
    cdAreaFrom: Optional[str] = Field(default=None, description="Area code from")
    cdAreaTo: Optional[str] = Field(default=None, description="Area code to")
    # ... see https://api.e-stat.go.jp/swagger-ui/e-statapi3.0.html#/

    # ページネーションパラメータ
    limit: int = Field(
        default=100000, description="Maximum number of records to fetch per request"
    )
    maximum_offset: Optional[int] = Field(
        default=None, description="Maximum number of records to fetch"
    )

    # 分類事項パラメータ（cat01-cat15）
    # 動的に生成されるため、Extraで受け入れる
    model_config = ConfigDict(extra="allow")

    @field_validator("statsDataId")
    @classmethod
    def validate_stats_data_id(cls, v: Union[str, List[str]]) -> Union[str, List[str]]:
        """Ensure statsDataId is valid."""
        if isinstance(v, list):
            if not v:
                raise ValueError("statsDataId list cannot be empty")
            for id_ in v:
                if not isinstance(id_, str) or not id_.strip():
                    raise ValueError(f"Invalid statsDataId: {id_}")
        elif isinstance(v, str):
            if not v.strip():
                raise ValueError("statsDataId cannot be empty")
        return v

`validate_stats_data_id(v)` `classmethod`

Ensure statsDataId is valid.

Source code in src/estat_api_dlt_helper/config/models.py

@field_validator("statsDataId")
@classmethod
def validate_stats_data_id(cls, v: Union[str, List[str]]) -> Union[str, List[str]]:
    """Ensure statsDataId is valid."""
    if isinstance(v, list):
        if not v:
            raise ValueError("statsDataId list cannot be empty")
        for id_ in v:
            if not isinstance(id_, str) or not id_.strip():
                raise ValueError(f"Invalid statsDataId: {id_}")
    elif isinstance(v, str):
        if not v.strip():
            raise ValueError("statsDataId cannot be empty")
    return v

DestinationConfig

dlt destination (データロード先) の設定クラスです。ロード先のDWH、データセット/テーブル名、書き込み戦略を含むdltの設定を定義します。

Bases: BaseModel

Configuration for DLT data destination.

Defines parameters for loading data to various destinations using DLT, including destination type, dataset/table names, and write strategies.

Attributes:

Name	Type	Description
`destination`	`Union[str, Any]`	DLT destination type or configuration object.
`dataset_name`	`str`	Target dataset/schema name.
`table_name`	`str`	Target table name.
`write_disposition`	`Literal['append', 'replace', 'merge']`	How to write data (append/replace/merge).
`primary_key`	`Optional[Union[str, List[str]]]`	Primary key columns for merge operations.

Source code in src/estat_api_dlt_helper/config/models.py

class DestinationConfig(BaseModel):
    """Configuration for DLT data destination.

    Defines parameters for loading data to various destinations using DLT,
    including destination type, dataset/table names, and write strategies.

    Attributes:
        destination: DLT destination type or configuration object.
        dataset_name: Target dataset/schema name.
        table_name: Target table name.
        write_disposition: How to write data (append/replace/merge).
        primary_key: Primary key columns for merge operations.
    """

    destination: Union[str, Any] = Field(
        ...,
        description="DLT destination configuration | e.g. 'bigquery', 'duckdb', 'filesystem', 'motherduck'",
    )
    dataset_name: str = Field(..., description="Dataset/schema name in the destination")
    table_name: str = Field(..., description="Table name in the destination")
    write_disposition: Literal["append", "replace", "merge"] = Field(
        default="merge", description="How to write data to the destination table"
    )
    primary_key: Optional[Union[str, List[str]]] = Field(
        default=["time", "area", "cat01"],
        description="Primary key column(s) for merge operations",
    )

    # DLT pipeline configuration
    pipeline_name: Optional[str] = Field(default=None, description="Name of the DLT pipeline")
    dev_mode: bool = Field(default=False, description="Enable DLT development mode")

    # Additional destination-specific configuration
    credentials: Optional[Dict[str, Any]] = Field(
        default=None, description="Destination-specific credentials"
    )
    extra_options: Optional[Dict[str, Any]] = Field(
        default=None, description="Additional destination-specific options"
    )

    @field_validator("primary_key")
    @classmethod
    def validate_primary_key(
        cls, v: Optional[Union[str, List[str]]], info
    ) -> Optional[Union[str, List[str]]]:
        """Validate primary key is provided for merge operations."""
        write_disposition = info.data.get("write_disposition")
        if write_disposition == "merge" and not v:
            raise ValueError(
                "primary_key must be specified when write_disposition is 'merge'"
            )
        return v

`validate_primary_key(v, info)` `classmethod`

Validate primary key is provided for merge operations.

Source code in src/estat_api_dlt_helper/config/models.py

@field_validator("primary_key")
@classmethod
def validate_primary_key(
    cls, v: Optional[Union[str, List[str]]], info
) -> Optional[Union[str, List[str]]]:
    """Validate primary key is provided for merge operations."""
    write_disposition = info.data.get("write_disposition")
    if write_disposition == "merge" and not v:
        raise ValueError(
            "primary_key must be specified when write_disposition is 'merge'"
        )
    return v

APIクライアント

EstatApiClient

e-Stat APIアクセス用のクライアントクラスです。政府統計のe-Stat API機能から統計データを取得するメソッドを提供し、API認証、リクエストフォーマット、レスポンス解析を処理します。

Client for accessing e-Stat API.

Provides methods to fetch statistical data from Japan's e-Stat API. Handles API authentication, request formatting, and response parsing. Uses dlt's requests Client for automatic retry, rate limiting, and connection pooling.

Attributes:

Name	Type	Description
`app_id`		e-Stat API application ID for authentication.
`base_url`		Base URL for API endpoints.
`timeout`		Request timeout in seconds.
`client`		HTTP client with retry and connection pooling.

Source code in src/estat_api_dlt_helper/api/client.py

class EstatApiClient:
    """Client for accessing e-Stat API.

    Provides methods to fetch statistical data from Japan's e-Stat API.
    Handles API authentication, request formatting, and response parsing.
    Uses dlt's requests Client for automatic retry, rate limiting, and
    connection pooling.

    Attributes:
        app_id: e-Stat API application ID for authentication.
        base_url: Base URL for API endpoints.
        timeout: Request timeout in seconds.
        client: HTTP client with retry and connection pooling.
    """

    def __init__(
        self,
        app_id: str,
        base_url: Optional[str] = None,
        timeout: int = 60,
    ):
        """Initialize e-Stat API client.

        Args:
            app_id: e-Stat API application ID
            base_url: Base URL for API (defaults to official endpoint)
            timeout: Request timeout in seconds
        """
        self.app_id = app_id
        self.base_url = base_url or ESTAT_ENDPOINTS["base_url"]
        self.timeout = timeout
        self.client = Client(request_timeout=timeout)
        self.default_headers = {"accept": "application/json"}

    def _make_request(
        self, endpoint: str, params: Dict[str, Any], **kwargs: Any
    ) -> Response:
        """Make HTTP request to e-Stat API.

        Args:
            endpoint: API endpoint name
            params: Query parameters
            **kwargs: Additional arguments for requests

        Returns:
            Response object
        """
        url = f"{self.base_url}{endpoint}"

        # Add appId to params
        params = {"appId": self.app_id, **params}

        logger.debug(f"Making request to {url} with params: {params}")

        response = self.client.get(
            url, params=params, headers=self.default_headers, **kwargs
        )

        return response

    def get_stats_data(
        self,
        stats_data_id: str,
        start_position: int = 1,
        limit: int = 100000,
        meta_get_flg: str = "Y",
        cnt_get_flg: str = "N",
        explanation_get_flg: str = "Y",
        annotation_get_flg: str = "Y",
        replace_sp_chars: str = "0",
        lang: str = "J",
        **additional_params: Any,
    ) -> Dict[str, Any]:
        """Get statistical data from e-Stat API.

        Args:
            stats_data_id: Statistical data ID
            start_position: Start position for data retrieval (1-based)
            limit: Maximum number of records to retrieve
            meta_get_flg: Whether to get metadata (Y/N)
            cnt_get_flg: Whether to get count only (Y/N)
            explanation_get_flg: Whether to get explanations (Y/N)
            annotation_get_flg: Whether to get annotations (Y/N)
            replace_sp_chars: Replace special characters (0: No, 1: Yes, 2: Remove)
            lang: Language (J: Japanese, E: English)
            **additional_params: Additional query parameters

        Returns:
            API response as dictionary
        """
        params = {
            "statsDataId": stats_data_id,
            "startPosition": start_position,
            "limit": limit,
            "metaGetFlg": meta_get_flg,
            "cntGetFlg": cnt_get_flg,
            "explanationGetFlg": explanation_get_flg,
            "annotationGetFlg": annotation_get_flg,
            "replaceSpChars": replace_sp_chars,
            "lang": lang,
            **additional_params,
        }

        response = self._make_request(ESTAT_ENDPOINTS["stats_data"], params)
        return response.json()

    def get_stats_data_generator(
        self, stats_data_id: str, limit_per_request: int = 100000, **kwargs: Any
    ) -> Generator[Dict[str, Any], None, None]:
        """Get statistical data as a generator for pagination.

        Args:
            stats_data_id: Statistical data ID
            limit_per_request: Number of records per request
            **kwargs: Additional parameters for get_stats_data

        Yields:
            Response data for each page
        """
        start_position = 1

        while True:
            response_data = self.get_stats_data(
                stats_data_id=stats_data_id,
                start_position=start_position,
                limit=limit_per_request,
                **kwargs,
            )

            # Extract data info
            stats_data = response_data.get("GET_STATS_DATA", {})
            statistical_data = stats_data.get("STATISTICAL_DATA", {})
            result_inf = statistical_data.get("RESULT_INF", {})

            # Get total number of records
            total_number = int(result_inf.get("TOTAL_NUMBER", 0))
            from_number = int(result_inf.get("FROM_NUMBER", 0))
            to_number = int(result_inf.get("TO_NUMBER", 0))

            logger.info(
                f"Retrieved records {from_number} to {to_number} of {total_number}"
            )

            yield response_data

            # Check if we've retrieved all records
            if to_number >= total_number:
                break

            # Update start position for next request
            start_position = to_number + 1

    def get_stats_list(
        self,
        search_word: Optional[str] = None,
        survey_years: Optional[str] = None,
        stats_code: Optional[str] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        """Get list of available statistics.

        Args:
            search_word: Search keyword
            survey_years: Survey years (YYYY or YYYYMM-YYYYMM)
            stats_code: Statistics code
            **kwargs: Additional query parameters

        Returns:
            API response as dictionary
        """
        params = {}

        if search_word:
            params["searchWord"] = search_word
        if survey_years:
            params["surveyYears"] = survey_years
        if stats_code:
            params["statsCode"] = stats_code

        params.update(kwargs)

        response = self._make_request(ESTAT_ENDPOINTS["stats_list"], params)
        return response.json()

    def close(self) -> None:
        """Close the underlying HTTP session."""
        self.client.session.close()

`init(app_id, base_url=None, timeout=60)`

Initialize e-Stat API client.

Parameters:

Name	Type	Description	Default
`app_id`	`str`	e-Stat API application ID	required
`base_url`	`Optional[str]`	Base URL for API (defaults to official endpoint)	`None`
`timeout`	`int`	Request timeout in seconds	`60`

Source code in src/estat_api_dlt_helper/api/client.py

def __init__(
    self,
    app_id: str,
    base_url: Optional[str] = None,
    timeout: int = 60,
):
    """Initialize e-Stat API client.

    Args:
        app_id: e-Stat API application ID
        base_url: Base URL for API (defaults to official endpoint)
        timeout: Request timeout in seconds
    """
    self.app_id = app_id
    self.base_url = base_url or ESTAT_ENDPOINTS["base_url"]
    self.timeout = timeout
    self.client = Client(request_timeout=timeout)
    self.default_headers = {"accept": "application/json"}

`close()`

Close the underlying HTTP session.

Source code in src/estat_api_dlt_helper/api/client.py

def close(self) -> None:
    """Close the underlying HTTP session."""
    self.client.session.close()

`get_stats_data(stats_data_id, start_position=1, limit=100000, meta_get_flg='Y', cnt_get_flg='N', explanation_get_flg='Y', annotation_get_flg='Y', replace_sp_chars='0', lang='J', **additional_params)`

Get statistical data from e-Stat API.

Parameters:

Name	Type	Description	Default
`stats_data_id`	`str`	Statistical data ID	required
`start_position`	`int`	Start position for data retrieval (1-based)	`1`
`limit`	`int`	Maximum number of records to retrieve	`100000`
`meta_get_flg`	`str`	Whether to get metadata (Y/N)	`'Y'`
`cnt_get_flg`	`str`	Whether to get count only (Y/N)	`'N'`
`explanation_get_flg`	`str`	Whether to get explanations (Y/N)	`'Y'`
`annotation_get_flg`	`str`	Whether to get annotations (Y/N)	`'Y'`
`replace_sp_chars`	`str`	Replace special characters (0: No, 1: Yes, 2: Remove)	`'0'`
`lang`	`str`	Language (J: Japanese, E: English)	`'J'`
`**additional_params`	`Any`	Additional query parameters	`{}`

Returns:

Type	Description
`Dict[str, Any]`	API response as dictionary

Source code in src/estat_api_dlt_helper/api/client.py

def get_stats_data(
    self,
    stats_data_id: str,
    start_position: int = 1,
    limit: int = 100000,
    meta_get_flg: str = "Y",
    cnt_get_flg: str = "N",
    explanation_get_flg: str = "Y",
    annotation_get_flg: str = "Y",
    replace_sp_chars: str = "0",
    lang: str = "J",
    **additional_params: Any,
) -> Dict[str, Any]:
    """Get statistical data from e-Stat API.

    Args:
        stats_data_id: Statistical data ID
        start_position: Start position for data retrieval (1-based)
        limit: Maximum number of records to retrieve
        meta_get_flg: Whether to get metadata (Y/N)
        cnt_get_flg: Whether to get count only (Y/N)
        explanation_get_flg: Whether to get explanations (Y/N)
        annotation_get_flg: Whether to get annotations (Y/N)
        replace_sp_chars: Replace special characters (0: No, 1: Yes, 2: Remove)
        lang: Language (J: Japanese, E: English)
        **additional_params: Additional query parameters

    Returns:
        API response as dictionary
    """
    params = {
        "statsDataId": stats_data_id,
        "startPosition": start_position,
        "limit": limit,
        "metaGetFlg": meta_get_flg,
        "cntGetFlg": cnt_get_flg,
        "explanationGetFlg": explanation_get_flg,
        "annotationGetFlg": annotation_get_flg,
        "replaceSpChars": replace_sp_chars,
        "lang": lang,
        **additional_params,
    }

    response = self._make_request(ESTAT_ENDPOINTS["stats_data"], params)
    return response.json()

`get_stats_data_generator(stats_data_id, limit_per_request=100000, **kwargs)`

Get statistical data as a generator for pagination.

Parameters:

Name	Type	Description	Default
`stats_data_id`	`str`	Statistical data ID	required
`limit_per_request`	`int`	Number of records per request	`100000`
`**kwargs`	`Any`	Additional parameters for get_stats_data	`{}`

Yields:

Type	Description
`Dict[str, Any]`	Response data for each page

Source code in src/estat_api_dlt_helper/api/client.py

def get_stats_data_generator(
    self, stats_data_id: str, limit_per_request: int = 100000, **kwargs: Any
) -> Generator[Dict[str, Any], None, None]:
    """Get statistical data as a generator for pagination.

    Args:
        stats_data_id: Statistical data ID
        limit_per_request: Number of records per request
        **kwargs: Additional parameters for get_stats_data

    Yields:
        Response data for each page
    """
    start_position = 1

    while True:
        response_data = self.get_stats_data(
            stats_data_id=stats_data_id,
            start_position=start_position,
            limit=limit_per_request,
            **kwargs,
        )

        # Extract data info
        stats_data = response_data.get("GET_STATS_DATA", {})
        statistical_data = stats_data.get("STATISTICAL_DATA", {})
        result_inf = statistical_data.get("RESULT_INF", {})

        # Get total number of records
        total_number = int(result_inf.get("TOTAL_NUMBER", 0))
        from_number = int(result_inf.get("FROM_NUMBER", 0))
        to_number = int(result_inf.get("TO_NUMBER", 0))

        logger.info(
            f"Retrieved records {from_number} to {to_number} of {total_number}"
        )

        yield response_data

        # Check if we've retrieved all records
        if to_number >= total_number:
            break

        # Update start position for next request
        start_position = to_number + 1

`get_stats_list(search_word=None, survey_years=None, stats_code=None, **kwargs)`

Get list of available statistics.

Parameters:

Name	Type	Description	Default
`search_word`	`Optional[str]`	Search keyword	`None`
`survey_years`	`Optional[str]`	Survey years (YYYY or YYYYMM-YYYYMM)	`None`
`stats_code`	`Optional[str]`	Statistics code	`None`
`**kwargs`	`Any`	Additional query parameters	`{}`

Returns:

Type	Description
`Dict[str, Any]`	API response as dictionary

Source code in src/estat_api_dlt_helper/api/client.py

def get_stats_list(
    self,
    search_word: Optional[str] = None,
    survey_years: Optional[str] = None,
    stats_code: Optional[str] = None,
    **kwargs: Any,
) -> Dict[str, Any]:
    """Get list of available statistics.

    Args:
        search_word: Search keyword
        survey_years: Survey years (YYYY or YYYYMM-YYYYMM)
        stats_code: Statistics code
        **kwargs: Additional query parameters

    Returns:
        API response as dictionary
    """
    params = {}

    if search_word:
        params["searchWord"] = search_word
    if survey_years:
        params["surveyYears"] = survey_years
    if stats_code:
        params["statsCode"] = stats_code

    params.update(kwargs)

    response = self._make_request(ESTAT_ENDPOINTS["stats_list"], params)
    return response.json()

データ解析

parse_response

e-Stat APIレスポンスを解析してArrow形式に変換する関数です。JSONレスポンスを受け取り、データ値と関連メタデータを含む構造化されたArrowテーブルを返します。

Parse e-Stat API response data and convert to Arrow table.

This is the main entry point for parsing e-Stat API responses. Takes the JSON response and returns a structured Arrow table with data values and associated metadata.

Parameters:

Name	Type	Description	Default
`data`	`Dict[str, Any]`	The complete JSON response from e-Stat API	required

Returns:

Type	Description
`Table`	pa.Table: Arrow table containing the parsed data with metadata

Raises:

Type	Description
`ValueError`	If required data sections are missing
`KeyError`	If expected keys are not found in the response

Source code in src/estat_api_dlt_helper/parser/response_parser.py

def parse_response(data: Dict[str, Any]) -> pa.Table:
    """
    Parse e-Stat API response data and convert to Arrow table.

    This is the main entry point for parsing e-Stat API responses.
    Takes the JSON response and returns a structured Arrow table with
    data values and associated metadata.

    Args:
        data: The complete JSON response from e-Stat API

    Returns:
        pa.Table: Arrow table containing the parsed data with metadata

    Raises:
        ValueError: If required data sections are missing
        KeyError: If expected keys are not found in the response
    """
    # Validate response structure
    if "GET_STATS_DATA" not in data:
        raise ValueError("Invalid response: missing GET_STATS_DATA section")

    stats_data = data["GET_STATS_DATA"]

    if "STATISTICAL_DATA" not in stats_data:
        raise ValueError("Invalid response: missing STATISTICAL_DATA section")

    statistical_data = stats_data["STATISTICAL_DATA"]

    # Check for required sections
    required_sections = ["TABLE_INF", "CLASS_INF", "DATA_INF"]
    missing_sections = [
        section for section in required_sections if section not in statistical_data
    ]

    if missing_sections:
        raise ValueError(
            f"Invalid response: missing required sections: {', '.join(missing_sections)}"
        )

    # Validate DATA_INF has VALUE
    if "VALUE" not in statistical_data["DATA_INF"]:
        raise ValueError("Invalid response: DATA_INF missing VALUE section")

    # Create processors
    metadata_processor = MetadataProcessor()
    arrow_converter = ArrowConverter(metadata_processor)

    # Convert to Arrow table
    return arrow_converter.convert_to_arrow(statistical_data)

データローダー関数

estat_table

単一のe-Stat API統計表をdlt resourceとして扱うための宣言的APIです。 stats_data_id を指定するだけでdlt resourceを生成でき、write_disposition、 primary_key、incremental などの設定もあわせて定義できます。

DLT resource for a single e-Stat statistical table.

`estat_table(stats_data_id, app_id=dlt.secrets.value, table_name=None, write_disposition='replace', primary_key=None, incremental=None, limit=_UNSET, maximum_offset=_UNSET, timeout=_UNSET, **api_params)`

Create a DLT resource for a single e-Stat statistical table.

Use this function directly when you need fine-grained control over write_disposition and primary_key per resource.

Parameters:

Name	Type	Description	Default
`stats_data_id`	`str`	Statistical table ID to fetch.	required
`app_id`	`str`	e-Stat API application ID. Resolved automatically from secrets.toml or environment variables when used inside a @dlt.source or @dlt.resource context.	`value`
`table_name`	`Optional[str]`	Resource/table name. Defaults to "estat_{stats_data_id}".	`None`
`write_disposition`	`str`	How to write data to destination.	`'replace'`
`primary_key`	`Optional[Union[str, List[str]]]`	Primary key column(s) for merge disposition.	`None`
`incremental`	`Optional[incremental[str]]`	Optional incremental loading configuration. Use dlt.sources.incremental("time", initial_value="0000000000") to enable incremental loading based on the time code column. When set, cdTimeFrom API parameter is automatically configured using the last loaded time code value. Best used with write_disposition="merge" or "append". Note that only new time points are detected; revisions to existing data are not captured.	`None`
`limit`	`int`	Maximum records per API request (pagination size).	`_UNSET`
`maximum_offset`	`Optional[int]`	Maximum total records to fetch. None for unlimited.	`_UNSET`
`timeout`	`int`	API request timeout in seconds.	`_UNSET`
`**api_params`	`Any`	Additional e-Stat API parameters (e.g., lang, cdTab, cdArea, cdTime, cdTimeFrom, cdTimeTo, cat01, etc.).	`{}`

Returns:

Type	Description
`DltResource`	DLT resource yielding PyArrow tables.

Example

import dlt
from estat_api_dlt_helper import estat_table

resource = estat_table(
    stats_data_id="0000020201",
    write_disposition="merge",
    primary_key=["time_code", "area_code"],
)
pipeline = dlt.pipeline(
    pipeline_name="estat",
    destination="duckdb",
    dataset_name="estat_data",
)
pipeline.run(resource)

Source code in src/estat_api_dlt_helper/loader/estat_table.py

def estat_table(
    stats_data_id: str,
    app_id: str = dlt.secrets.value,
    table_name: Optional[str] = None,
    write_disposition: str = "replace",
    primary_key: Optional[Union[str, List[str]]] = None,
    incremental: Optional[dlt_incremental[str]] = None,
    limit: int = _UNSET,  # type: ignore[assignment]  # sentinel to detect explicit args
    maximum_offset: Optional[int] = _UNSET,  # type: ignore[assignment]  # sentinel to detect explicit args
    timeout: int = _UNSET,  # type: ignore[assignment]  # sentinel to detect explicit args
    **api_params: Any,
) -> DltResource:
    """Create a DLT resource for a single e-Stat statistical table.

    Use this function directly when you need fine-grained control over
    write_disposition and primary_key per resource.

    Args:
        stats_data_id: Statistical table ID to fetch.
        app_id: e-Stat API application ID. Resolved automatically from
            secrets.toml or environment variables when used inside a
            @dlt.source or @dlt.resource context.
        table_name: Resource/table name. Defaults to "estat_{stats_data_id}".
        write_disposition: How to write data to destination.
        primary_key: Primary key column(s) for merge disposition.
        incremental: Optional incremental loading configuration.
            Use dlt.sources.incremental("time", initial_value="0000000000")
            to enable incremental loading based on the time code column.
            When set, cdTimeFrom API parameter is automatically configured
            using the last loaded time code value. Best used with
            write_disposition="merge" or "append". Note that only new time
            points are detected; revisions to existing data are not captured.
        limit: Maximum records per API request (pagination size).
        maximum_offset: Maximum total records to fetch. None for unlimited.
        timeout: API request timeout in seconds.
        **api_params: Additional e-Stat API parameters (e.g., lang, cdTab,
            cdArea, cdTime, cdTimeFrom, cdTimeTo, cat01, etc.).

    Returns:
        DLT resource yielding PyArrow tables.

    Example:
        ```python
        import dlt
        from estat_api_dlt_helper import estat_table

        resource = estat_table(
            stats_data_id="0000020201",
            write_disposition="merge",
            primary_key=["time_code", "area_code"],
        )
        pipeline = dlt.pipeline(
            pipeline_name="estat",
            destination="duckdb",
            dataset_name="estat_data",
        )
        pipeline.run(resource)
        ```
    """
    _table_explicit_args: set[str] = set()
    if limit is not _UNSET:
        _table_explicit_args.add("limit")
    else:
        limit = 100000
    if maximum_offset is not _UNSET:
        _table_explicit_args.add("maximum_offset")
    else:
        maximum_offset = None
    if timeout is not _UNSET:
        _table_explicit_args.add("timeout")
    else:
        timeout = 60

    if not stats_data_id or not stats_data_id.strip():
        raise ValueError("stats_data_id must not be empty")

    resource_name = table_name or f"estat_{stats_data_id}"

    merged_params = {**_DEFAULT_API_PARAMS, **api_params}
    params = _build_api_params(**merged_params)

    resource_config: Dict[str, Any] = {
        "name": resource_name,
        "write_disposition": write_disposition,
        "schema_contract": {
            "tables": "evolve",
            "columns": "evolve",
            "data_type": "freeze",
        },
    }
    if primary_key is not None:
        resource_config["primary_key"] = primary_key

    @dlt.resource(**resource_config)  # type: ignore[arg-type]
    def _estat_data(
        app_id: str = app_id,
        time_incremental: Optional[dlt_incremental[str]] = incremental,
        limit: int = limit,
        maximum_offset: Optional[int] = maximum_offset,
        timeout: int = timeout,
    ) -> Generator[pa.Table, None, None]:
        request_params = dict(params)
        if time_incremental is not None and time_incremental.start_value is not None:
            request_params["cdTimeFrom"] = time_incremental.start_value

        client = EstatApiClient(app_id=app_id, timeout=timeout)
        try:
            yield from _fetch_estat_data(
                client=client,
                stats_data_id=stats_data_id,
                params=request_params,
                limit=limit,
                maximum_offset=maximum_offset,
            )
        finally:
            client.close()

    _estat_data._table_explicit_args = _table_explicit_args  # type: ignore[attr-defined]
    return _estat_data

estat_source

複数のe-Stat API統計表をdlt sourceとしてまとめて扱うための宣言的APIです。 stats_data_ids に単一ID、IDリスト、{resource_name: stats_data_id} 形式の辞書を渡せます。また、tables に estat_table() のリストを渡すことで、リソースごとの個別設定も可能です。

DLT source for e-Stat API data.

`estat_source(stats_data_ids=None, tables=None, app_id=dlt.secrets.value, write_disposition='replace', primary_key=None, incremental=None, limit=100000, maximum_offset=None, timeout=60, **api_params)`

Create a DLT source for e-Stat API statistical data.

Supports two modes: - stats_data_ids: Each ID becomes a separate DLT resource via estat_table, sharing write_disposition/primary_key settings. - tables: Pass pre-configured estat_table resources directly for per-resource control.

Parameters:

Name	Type	Description	Default
`stats_data_ids`	`Union[str, List[str], Dict[str, str], None]`	Statistical table ID(s) to fetch. Accepts: - str: single ID (resource name: "estat_{id}") - List[str]: multiple IDs (resource names: "estat_{id}") - Dict[str, str]: {resource_name: stats_data_id} for custom names	`None`
`tables`	`Optional[List[DltResource]]`	Pre-configured DltResource list (from estat_table). When provided, write_disposition/primary_key are ignored as each resource carries its own settings. app_id/limit/maximum_offset/timeout are propagated to each table via bind().	`None`
`app_id`	`str`	e-Stat API application ID. Resolved automatically from secrets.toml ([sources.estat] app_id) or environment variable (SOURCES__ESTAT__APP_ID) if not provided.	`value`
`write_disposition`	`str`	How to write data to destination. Applied to all resources when using stats_data_ids.	`'replace'`
`primary_key`	`Optional[Union[str, List[str]]]`	Primary key column(s) for merge disposition. Applied to all resources when using stats_data_ids.	`None`
`incremental`	`Optional[incremental[str]]`	Optional incremental loading configuration. Applied to all resources when using stats_data_ids mode. Ignored when using tables mode (each table carries its own config).	`None`
`limit`	`int`	Maximum records per API request (pagination size).	`100000`
`maximum_offset`	`Optional[int]`	Maximum total records to fetch. None for unlimited.	`None`
`timeout`	`int`	API request timeout in seconds.	`60`
`**api_params`	`Any`	Additional e-Stat API parameters passed directly to the API request (e.g., lang, lvTab, cdTab, cdTime, cdArea, cdTimeFrom, cdTimeTo, metaGetFlg, cntGetFlg, replaceSpChars, cat01, etc.).	`{}`

Returns:

Type	Description
`Iterable[DltResource]`	DltSource with one resource per stats_data_id
`Iterable[DltResource]`	(the @dlt.source decorator wraps the generator into a DltSource).

Raises:

Type	Description
`ValueError`	If both stats_data_ids and tables are provided, if neither is provided, if tables is an empty list, or if tables is used with write_disposition/primary_key/ incremental/api_params arguments.

Example

import dlt
from estat_api_dlt_helper import estat_source, estat_table

# Simple: multiple tables with shared settings
source = estat_source(
    stats_data_ids=["0000020201", "0004028584"],
    write_disposition="merge",
    primary_key=["time_code", "area_code"],
)

# Advanced: per-resource settings
source = estat_source(
    tables=[
        estat_table(stats_data_id="0000020201", table_name="pop",
                    write_disposition="merge", primary_key=["time_code"]),
        estat_table(stats_data_id="0004028584", table_name="gdp",
                    write_disposition="replace"),
    ],
)

pipeline = dlt.pipeline(
    pipeline_name="estat",
    destination="duckdb",
    dataset_name="estat_data",
)
pipeline.run(source)

Source code in src/estat_api_dlt_helper/loader/estat_source.py

@dlt.source(name="estat")
def estat_source(
    stats_data_ids: Union[str, List[str], Dict[str, str], None] = None,
    tables: Optional[List[DltResource]] = None,
    app_id: str = dlt.secrets.value,
    write_disposition: str = "replace",
    primary_key: Optional[Union[str, List[str]]] = None,
    incremental: Optional[dlt_incremental[str]] = None,
    limit: int = 100000,
    maximum_offset: Optional[int] = None,
    timeout: int = 60,
    **api_params: Any,
) -> Iterable[DltResource]:
    """Create a DLT source for e-Stat API statistical data.

    Supports two modes:
    - stats_data_ids: Each ID becomes a separate DLT resource via estat_table,
      sharing write_disposition/primary_key settings.
    - tables: Pass pre-configured estat_table resources directly for
      per-resource control.

    Args:
        stats_data_ids: Statistical table ID(s) to fetch. Accepts:
            - str: single ID (resource name: "estat_{id}")
            - List[str]: multiple IDs (resource names: "estat_{id}")
            - Dict[str, str]: {resource_name: stats_data_id} for custom names
        tables: Pre-configured DltResource list (from estat_table).
            When provided, write_disposition/primary_key are ignored
            as each resource carries its own settings.
            app_id/limit/maximum_offset/timeout are propagated to each
            table via bind().
        app_id: e-Stat API application ID. Resolved automatically from
            secrets.toml ([sources.estat] app_id) or environment variable
            (SOURCES__ESTAT__APP_ID) if not provided.
        write_disposition: How to write data to destination.
            Applied to all resources when using stats_data_ids.
        primary_key: Primary key column(s) for merge disposition.
            Applied to all resources when using stats_data_ids.
        incremental: Optional incremental loading configuration.
            Applied to all resources when using stats_data_ids mode.
            Ignored when using tables mode (each table carries its own config).
        limit: Maximum records per API request (pagination size).
        maximum_offset: Maximum total records to fetch. None for unlimited.
        timeout: API request timeout in seconds.
        **api_params: Additional e-Stat API parameters passed directly to the
            API request (e.g., lang, lvTab, cdTab, cdTime, cdArea, cdTimeFrom,
            cdTimeTo, metaGetFlg, cntGetFlg, replaceSpChars, cat01, etc.).

    Returns:
        DltSource with one resource per stats_data_id
        (the @dlt.source decorator wraps the generator into a DltSource).

    Raises:
        ValueError: If both stats_data_ids and tables are provided,
            if neither is provided, if tables is an empty list,
            or if tables is used with write_disposition/primary_key/
            incremental/api_params arguments.

    Example:
        ```python
        import dlt
        from estat_api_dlt_helper import estat_source, estat_table

        # Simple: multiple tables with shared settings
        source = estat_source(
            stats_data_ids=["0000020201", "0004028584"],
            write_disposition="merge",
            primary_key=["time_code", "area_code"],
        )

        # Advanced: per-resource settings
        source = estat_source(
            tables=[
                estat_table(stats_data_id="0000020201", table_name="pop",
                            write_disposition="merge", primary_key=["time_code"]),
                estat_table(stats_data_id="0004028584", table_name="gdp",
                            write_disposition="replace"),
            ],
        )

        pipeline = dlt.pipeline(
            pipeline_name="estat",
            destination="duckdb",
            dataset_name="estat_data",
        )
        pipeline.run(source)
        ```
    """
    if tables is not None and stats_data_ids is not None:
        raise ValueError(
            "Cannot specify both stats_data_ids and tables. Use one or the other."
        )
    if tables is None and stats_data_ids is None:
        raise ValueError("Either stats_data_ids or tables must be provided.")
    if tables is not None and not tables:
        raise ValueError("tables must not be empty")

    if tables is not None:
        conflicting = {
            # True if non-default value is specified
            "write_disposition": write_disposition != "replace",
            "primary_key": primary_key is not None,
            "incremental": incremental is not None,
            "api_params": bool(api_params),  # True if not empty
        }
        found = [k for k, v in conflicting.items() if v]
        if found:
            raise ValueError(
                f"When using 'tables' mode, the following arguments are not "
                f"supported because each table carries its own settings: "
                f"{', '.join(found)}. "
                f"Remove these arguments or configure them on each estat_table() call."
            )
        for table in tables:
            table_explicit = getattr(table, "_table_explicit_args", set())
            bind_kwargs: Dict[str, Any] = {"app_id": app_id}
            if "limit" not in table_explicit:
                bind_kwargs["limit"] = limit
            if "maximum_offset" not in table_explicit:
                bind_kwargs["maximum_offset"] = maximum_offset
            if "timeout" not in table_explicit:
                bind_kwargs["timeout"] = timeout
            table.bind(**bind_kwargs)
            yield table
        return

    assert stats_data_ids is not None  # guaranteed by validation above
    id_map = _normalize_stats_data_ids(stats_data_ids)
    for resource_name, stats_data_id in id_map.items():
        yield estat_table(
            stats_data_id=stats_data_id,
            app_id=app_id,
            table_name=resource_name,
            write_disposition=write_disposition,
            primary_key=primary_key,
            incremental=incremental,
            limit=limit,
            maximum_offset=maximum_offset,
            timeout=timeout,
            **api_params,
        )

load_estat_data

e-Stat APIデータを指定されたデスティネーションにロードする便利な関数です。提供された設定でdltパイプラインを作成して実行します。

Load e-Stat API data to the specified destination using DLT.

This is a convenience function that creates and runs a DLT pipeline with the provided configuration.

Parameters:

Name	Type	Description	Default
`config`	`EstatDltConfig`	Configuration for e-Stat API source and DLT destination	required
`credentials`	`Optional[Dict[str, Any]]`	Optional credentials to override destination credentials	`None`
`**kwargs`	`Any`	Additional arguments passed to pipeline.run()	`{}`

Returns:

Type	Description
`Any`	LoadInfo object containing information about the load operation

Example

from estat_api_dlt_helper import EstatDltConfig, load_estat_data

config = {
    "source": {
        "app_id": "YOUR_API_KEY",
        "statsDataId": "0000020211",
        "limit": 10
    },
    "destination": {
        "destination": "duckdb",
        "dataset_name": "demo",
        "table_name": "demo",
        "write_disposition": "merge",
        "primary_key": ["time", "area", "cat01"]
    }
}

config = EstatDltConfig(**config)
info = load_estat_data(config)
print(info)

Source code in src/estat_api_dlt_helper/loader/load_manager.py

def load_estat_data(
    config: EstatDltConfig,
    *,
    credentials: Optional[Dict[str, Any]] = None,
    **kwargs: Any,
) -> Any:  # dlt.common.pipeline.LoadInfo
    """
    Load e-Stat API data to the specified destination using DLT.

    This is a convenience function that creates and runs a DLT pipeline
    with the provided configuration.

    Args:
        config: Configuration for e-Stat API source and DLT destination
        credentials: Optional credentials to override destination credentials
        **kwargs: Additional arguments passed to pipeline.run()

    Returns:
        LoadInfo object containing information about the load operation

    Example:
        ```python
        from estat_api_dlt_helper import EstatDltConfig, load_estat_data

        config = {
            "source": {
                "app_id": "YOUR_API_KEY",
                "statsDataId": "0000020211",
                "limit": 10
            },
            "destination": {
                "destination": "duckdb",
                "dataset_name": "demo",
                "table_name": "demo",
                "write_disposition": "merge",
                "primary_key": ["time", "area", "cat01"]
            }
        }

        config = EstatDltConfig(**config)
        info = load_estat_data(config)
        print(info)
        ```
    """
    logger.info("Starting e-Stat data load process")

    try:
        # Override credentials if provided
        if credentials:
            config.destination.credentials = credentials

        # Create the resource
        logger.debug("Creating e-Stat resource")
        resource = create_estat_resource(config)

        # Create the pipeline
        logger.debug("Creating DLT pipeline")
        pipeline = create_estat_pipeline(config)

        # Run the pipeline
        logger.info(
            f"Running pipeline for stats_data_id: {config.source.statsDataId} "
            f"to {config.destination.destination}/{config.destination.dataset_name}/"
            f"{config.destination.table_name}"
        )

        info = pipeline.run(resource, **kwargs)

        # Log results
        logger.info(f"Load completed: {info}")

        return info

    except Exception as e:
        logger.error(f"Error during data load: {e}")
        raise

create_estat_resource

e-Stat APIデータ用のdltリソースを作成する関数です。設定に基づいてe-Stat APIからデータを取得するカスタマイズ可能なdltリソースを作成します。

Create a DLT resource for e-Stat API data.

This function creates a customizable DLT resource that fetches data from the e-Stat API based on the provided configuration.

Parameters:

Name	Type	Description	Default
`config`	`EstatDltConfig`	Configuration for e-Stat API source and destination	required
`name`	`Optional[str]`	Resource name (defaults to table_name from config)	`None`
`primary_key`	`Optional[Any]`	Primary key columns (overrides config if provided)	`None`
`write_disposition`	`Optional[str]`	Write disposition (overrides config if provided)	`None`
`columns`	`Optional[Any]`	Column definitions for the resource	`None`
`table_format`	`Optional[str]`	Table format for certain destinations	`None`
`file_format`	`Optional[str]`	File format for filesystem destinations	`None`
`schema_contract`	`Optional[Any]`	Schema contract settings	`None`
`table_name`	`Optional[Callable[[Any], str]]`	Callable to generate dynamic table names	`None`
`max_table_nesting`	`Optional[int]`	Maximum nesting level for nested data	`None`
`selected`	`Optional[bool]`	Whether this resource is selected for loading	`None`
`merge_key`	`Optional[Any]`	Merge key for merge operations	`None`
`parallelized`	`Optional[bool]`	Whether to parallelize this resource	`None`
`**resource_kwargs`	`Any`	Additional keyword arguments for dlt.resource	`{}`

Returns:

Name	Type	Description
`DltResource`	`DltResource`	Configured DLT resource for e-Stat API data

Example

from estat_api_dlt_helper import EstatDltConfig, create_estat_resource

config = EstatDltConfig(...)
resource = create_estat_resource(config)

# Customize the resource
resource = create_estat_resource(
    config,
    name="custom_stats",
    columns={"time": {"data_type": "timestamp"}},
    selected=True
)

Source code in src/estat_api_dlt_helper/loader/dlt_resource.py

def create_estat_resource(
    config: EstatDltConfig,
    *,
    name: Optional[str] = None,
    primary_key: Optional[Any] = None,
    write_disposition: Optional[str] = None,
    columns: Optional[Any] = None,
    table_format: Optional[str] = None,
    file_format: Optional[str] = None,
    schema_contract: Optional[Any] = None,
    table_name: Optional[Callable[[Any], str]] = None,
    max_table_nesting: Optional[int] = None,
    selected: Optional[bool] = None,
    merge_key: Optional[Any] = None,
    parallelized: Optional[bool] = None,
    **resource_kwargs: Any,
) -> DltResource:
    """
    Create a DLT resource for e-Stat API data.

    This function creates a customizable DLT resource that fetches data
    from the e-Stat API based on the provided configuration.

    Args:
        config: Configuration for e-Stat API source and destination
        name: Resource name (defaults to table_name from config)
        primary_key: Primary key columns (overrides config if provided)
        write_disposition: Write disposition (overrides config if provided)
        columns: Column definitions for the resource
        table_format: Table format for certain destinations
        file_format: File format for filesystem destinations
        schema_contract: Schema contract settings
        table_name: Callable to generate dynamic table names
        max_table_nesting: Maximum nesting level for nested data
        selected: Whether this resource is selected for loading
        merge_key: Merge key for merge operations
        parallelized: Whether to parallelize this resource
        **resource_kwargs: Additional keyword arguments for dlt.resource

    Returns:
        DltResource: Configured DLT resource for e-Stat API data

    Example:
        ```python
        from estat_api_dlt_helper import EstatDltConfig, create_estat_resource

        config = EstatDltConfig(...)
        resource = create_estat_resource(config)

        # Customize the resource
        resource = create_estat_resource(
            config,
            name="custom_stats",
            columns={"time": {"data_type": "timestamp"}},
            selected=True
        )
        ```
    """
    # Prepare API parameters
    api_params = _create_api_params(config)

    # Get stats data IDs (ensure it's a list)
    stats_data_ids = config.source.statsDataId
    if isinstance(stats_data_ids, str):
        stats_data_ids = [stats_data_ids]

    # Prepare resource configuration
    resource_config: Dict[str, Any] = {
        "name": name or config.destination.table_name,
        "write_disposition": write_disposition or config.destination.write_disposition,
        # Allow schema evolution for handling different metadata structures
        "schema_contract": schema_contract
        or {
            "tables": "evolve",
            "columns": "evolve",  # Allow new columns like parent_code in time_metadata
            "data_type": "freeze",  # Keep data types consistent
        },
    }

    # Add primary key for merge disposition
    if primary_key is not None:
        resource_config["primary_key"] = primary_key
    elif (
        config.destination.write_disposition == "merge"
        and config.destination.primary_key
    ):
        pk = config.destination.primary_key
        if isinstance(pk, str):
            pk = [pk]
        resource_config["primary_key"] = pk

    # Add optional resource parameters
    optional_params = {
        "columns": columns,
        "table_format": table_format,
        "file_format": file_format,
        "schema_contract": schema_contract,
        "table_name": table_name,
        "max_table_nesting": max_table_nesting,
        "selected": selected,
        "merge_key": merge_key,
        "parallelized": parallelized,
    }

    for key, value in optional_params.items():
        if value is not None:
            resource_config[key] = value

    # Add any additional resource kwargs
    resource_config.update(resource_kwargs)

    @dlt.resource(**resource_config)  # type: ignore
    def estat_data() -> Generator[pa.Table, None, None]:
        """Generator function for e-Stat data."""
        client_kwargs: Dict[str, Any] = {"app_id": config.source.app_id}
        if config.timeout is not None:
            client_kwargs["timeout"] = config.timeout
        client = EstatApiClient(**client_kwargs)

        try:
            # Process each stats data ID
            for stats_data_id in stats_data_ids:
                yield from _fetch_estat_data(
                    client=client,
                    stats_data_id=stats_data_id,
                    params=api_params,
                    limit=config.source.limit,
                    maximum_offset=config.source.maximum_offset,
                )
        finally:
            client.close()

    return estat_data()

create_estat_pipeline

e-Stat APIデータロード用のdltパイプラインを作成する関数です。提供された設定に基づいて指定されたデスティネーション用に構成されたカスタマイズ可能なdltパイプラインを作成します。

Create a DLT pipeline for e-Stat API data loading.

This function creates a customizable DLT pipeline configured for the specified destination based on the provided configuration.

Parameters:

Name	Type	Description	Default
`config`	`EstatDltConfig`	Configuration for e-Stat API source and destination	required
`pipeline_name`	`Optional[str]`	Name of the pipeline (overrides config if provided)	`None`
`pipelines_dir`	`Optional[str]`	Directory to store pipeline state	`None`
`dataset_name`	`Optional[str]`	Dataset name in destination (overrides config if provided)	`None`
`import_schema_path`	`Optional[str]`	Path to import schema from	`None`
`export_schema_path`	`Optional[str]`	Path to export schema to	`None`
`dev_mode`	`Optional[bool]`	Development mode (overrides config if provided)	`None`
`refresh`	`Optional[str]`	Schema refresh mode	`None`
`progress`	`Optional[str]`	Progress reporting configuration	`None`
`destination`	`Optional[Any]`	DLT destination (constructed from config if not provided)	`None`
`staging`	`Optional[Any]`	Staging destination for certain loaders	`None`
`**pipeline_kwargs`	`Any`	Additional keyword arguments for dlt.pipeline	`{}`

Returns:

Name	Type	Description
`Pipeline`	`Pipeline`	Configured DLT pipeline

Example

from estat_api_dlt_helper import EstatDltConfig, create_estat_pipeline

config = EstatDltConfig(...)
pipeline = create_estat_pipeline(config)

# Customize the pipeline
pipeline = create_estat_pipeline(
    config,
    pipeline_name="custom_estat_pipeline",
    dev_mode=True,
    progress="log"
)

Source code in src/estat_api_dlt_helper/loader/dlt_pipeline.py

def create_estat_pipeline(
    config: EstatDltConfig,
    *,
    pipeline_name: Optional[str] = None,
    pipelines_dir: Optional[str] = None,
    dataset_name: Optional[str] = None,
    import_schema_path: Optional[str] = None,
    export_schema_path: Optional[str] = None,
    dev_mode: Optional[bool] = None,
    refresh: Optional[str] = None,
    progress: Optional[str] = None,
    destination: Optional[Any] = None,
    staging: Optional[Any] = None,
    **pipeline_kwargs: Any,
) -> Pipeline:
    """
    Create a DLT pipeline for e-Stat API data loading.

    This function creates a customizable DLT pipeline configured for
    the specified destination based on the provided configuration.

    Args:
        config: Configuration for e-Stat API source and destination
        pipeline_name: Name of the pipeline (overrides config if provided)
        pipelines_dir: Directory to store pipeline state
        dataset_name: Dataset name in destination (overrides config if provided)
        import_schema_path: Path to import schema from
        export_schema_path: Path to export schema to
        dev_mode: Development mode (overrides config if provided)
        refresh: Schema refresh mode
        progress: Progress reporting configuration
        destination: DLT destination (constructed from config if not provided)
        staging: Staging destination for certain loaders
        **pipeline_kwargs: Additional keyword arguments for dlt.pipeline

    Returns:
        Pipeline: Configured DLT pipeline

    Example:
        ```python
        from estat_api_dlt_helper import EstatDltConfig, create_estat_pipeline

        config = EstatDltConfig(...)
        pipeline = create_estat_pipeline(config)

        # Customize the pipeline
        pipeline = create_estat_pipeline(
            config,
            pipeline_name="custom_estat_pipeline",
            dev_mode=True,
            progress="log"
        )
        ```
    """
    # Determine pipeline name
    name = pipeline_name or config.destination.pipeline_name
    if not name:
        # Generate default pipeline name
        stats_id = config.source.statsDataId
        if isinstance(stats_id, list):
            stats_id = "_".join(stats_id[:3])  # Limit to first 3 IDs
            if len(config.source.statsDataId) > 3:
                stats_id += "_etc"
        else:
            stats_id = stats_id

        name = f"estat_{config.destination.dataset_name}_{stats_id}"

    # Prepare destination configuration
    dest = destination or config.destination.destination

    # Handle destination-specific configurations
    if isinstance(dest, str):
        # String destination name
        if config.destination.credentials:
            # Create destination with credentials
            # For now, just use the string destination name
            # DLT will handle the destination creation internally
            pass

    # Prepare pipeline configuration
    pipeline_config = {
        "pipeline_name": name,
        "destination": dest,
        "dataset_name": dataset_name or config.destination.dataset_name,
    }

    # Add optional parameters with proper defaults
    if dev_mode is not None:
        pipeline_config["dev_mode"] = dev_mode
    elif config.destination.dev_mode is not None:
        pipeline_config["dev_mode"] = config.destination.dev_mode

    # Add other optional parameters
    optional_params = {
        "pipelines_dir": pipelines_dir,
        "import_schema_path": import_schema_path,
        "export_schema_path": export_schema_path,
        "refresh": refresh,
        "progress": progress,
        "staging": staging,
    }

    for key, value in optional_params.items():
        if value is not None:
            pipeline_config[key] = value

    # Add any additional pipeline kwargs
    pipeline_config.update(pipeline_kwargs)

    # Create and return the pipeline
    logger.info(
        f"Creating pipeline '{name}' for destination '{dest}' "
        f"with dataset '{pipeline_config['dataset_name']}'"
    )

    return dlt.pipeline(**pipeline_config)

create_estat_source

複数のe-Stat API統計表を一括でロードするためのdltソースを作成する関数です。EstatDltConfigのリストを渡すことで、各設定に対応するリソースをまとめたdlt sourceを生成します。

Create a DLT source for multiple e-Stat API tables.

This function creates a DLT source containing one resource per config, using create_estat_resource() internally.

Parameters:

Name	Type	Description	Default
`configs`	`List[EstatDltConfig]`	List of EstatDltConfig, one per resource.	required
`**source_kwargs`	`Any`	Additional keyword arguments for dlt.source	`{}`

Returns:

Type	Description
`DltSource`	DLT source with one resource per config.

Example

from estat_api_dlt_helper import EstatDltConfig, create_estat_source

configs = [
    EstatDltConfig(
        source={"app_id": "YOUR_KEY", "statsDataId": "0000020201"},
        destination={"destination": "duckdb", "dataset_name": "estat", "table_name": "pop"},
    ),
    EstatDltConfig(
        source={"app_id": "YOUR_KEY", "statsDataId": "0004028584"},
        destination={"destination": "duckdb", "dataset_name": "estat", "table_name": "gdp"},
    ),
]
source = create_estat_source(configs)

Source code in src/estat_api_dlt_helper/loader/dlt_source.py

def create_estat_source(
    configs: List[EstatDltConfig],
    **source_kwargs: Any,
) -> DltSource:
    """
    Create a DLT source for multiple e-Stat API tables.

    This function creates a DLT source containing one resource per config,
    using create_estat_resource() internally.

    Args:
        configs: List of EstatDltConfig, one per resource.
        **source_kwargs: Additional keyword arguments for dlt.source

    Returns:
        DLT source with one resource per config.

    Example:
        ```python
        from estat_api_dlt_helper import EstatDltConfig, create_estat_source

        configs = [
            EstatDltConfig(
                source={"app_id": "YOUR_KEY", "statsDataId": "0000020201"},
                destination={"destination": "duckdb", "dataset_name": "estat", "table_name": "pop"},
            ),
            EstatDltConfig(
                source={"app_id": "YOUR_KEY", "statsDataId": "0004028584"},
                destination={"destination": "duckdb", "dataset_name": "estat", "table_name": "gdp"},
            ),
        ]
        source = create_estat_source(configs)
        ```
    """
    if not configs:
        raise ValueError("configs must not be empty")

    table_names = [config.destination.table_name for config in configs]
    duplicates = [name for name in table_names if table_names.count(name) > 1]
    if duplicates:
        raise ValueError(
            f"Duplicate table names found: {sorted(set(duplicates))}"
        )

    source_config: Dict[str, Any] = dict(source_kwargs)

    @dlt.source(**source_config)
    def estat_source() -> Sequence[Any]:
        resources: List[Any] = []
        for config in configs:
            resource = create_estat_resource(config)
            resources.append(resource)
        return resources

    return estat_source()

create_unified_estat_resource

異なるメタデータ構造を持つ複数のe-Stat APIデータセットを統一スキーマで処理するdltリソースを作成する関数です。複数のstatsDataIdを読み込む際に発生する「Schema at index X was different」エラーを防ぎます。

Create a DLT resource for e-Stat API data using unified schema.

This function creates a DLT resource that handles multiple e-Stat datasets with varying metadata structures by using a unified schema approach. This prevents PyArrow "Schema at index X was different" errors when loading multiple statsDataIds.

The unified schema includes all possible fields from all datasets, with missing fields automatically set to None. Column order from the original data is preserved.

Parameters:

Name	Type	Description	Default
`config`	`EstatDltConfig`	Configuration for e-Stat API source and destination	required
`name`	`Optional[str]`	Resource name (defaults to table_name from config)	`None`
`primary_key`	`Optional[Any]`	Primary key columns (overrides config if provided)	`None`
`write_disposition`	`Optional[str]`	Write disposition (overrides config if provided)	`None`
`columns`	`Optional[Any]`	Column definitions for the resource	`None`
`table_format`	`Optional[str]`	Table format for certain destinations	`None`
`file_format`	`Optional[str]`	File format for filesystem destinations	`None`
`schema_contract`	`Optional[Any]`	Schema contract settings (defaults to evolve mode)	`None`
`table_name`	`Optional[Callable[[Any], str]]`	Callable to generate dynamic table names	`None`
`max_table_nesting`	`Optional[int]`	Maximum nesting level for nested data	`None`
`selected`	`Optional[bool]`	Whether this resource is selected for loading	`None`
`merge_key`	`Optional[Any]`	Merge key for merge operations	`None`
`parallelized`	`Optional[bool]`	Whether to parallelize this resource	`None`
`**resource_kwargs`	`Any`	Additional keyword arguments for dlt.resource	`{}`

Returns:

Type	Description
`Any`	dlt.Resource: Configured DLT resource with unified schema support

Example

from estat_api_dlt_helper import EstatDltConfig
from estat_api_dlt_helper.loader.unified_schema_resource import create_unified_estat_resource

# Configuration with multiple statsDataIds that have different schemas
config = EstatDltConfig(
    source={
        "statsDataId": ["0004028473", "0004028474", "0004028475"],
        "app_id": "YOUR_APP_ID"
    },
    destination={
        "table_name": "unified_stats",
        "dataset_name": "estat_data"
    }
)

# Create resource with unified schema
resource = create_unified_estat_resource(config)

# Run in pipeline without schema errors
import dlt
pipeline = dlt.pipeline(
    pipeline_name="estat_unified",
    destination="duckdb",
    dataset_name="estat_data"
)
pipeline.run(resource)

Note

This resource is recommended when loading multiple statsDataIds that may have different metadata structures (e.g., some have parent_code in time_metadata while others don't).

Source code in src/estat_api_dlt_helper/loader/unified_schema_resource.py

def create_unified_estat_resource(
    config: EstatDltConfig,
    *,
    name: Optional[str] = None,
    primary_key: Optional[Any] = None,
    write_disposition: Optional[str] = None,
    columns: Optional[Any] = None,
    table_format: Optional[str] = None,
    file_format: Optional[str] = None,
    schema_contract: Optional[Any] = None,
    table_name: Optional[Callable[[Any], str]] = None,
    max_table_nesting: Optional[int] = None,
    selected: Optional[bool] = None,
    merge_key: Optional[Any] = None,
    parallelized: Optional[bool] = None,
    **resource_kwargs: Any,
) -> Any:  # dlt.Resource
    """
    Create a DLT resource for e-Stat API data using unified schema.

    This function creates a DLT resource that handles multiple e-Stat datasets
    with varying metadata structures by using a unified schema approach. This
    prevents PyArrow "Schema at index X was different" errors when loading
    multiple statsDataIds.

    The unified schema includes all possible fields from all datasets, with
    missing fields automatically set to None. Column order from the original
    data is preserved.

    Args:
        config: Configuration for e-Stat API source and destination
        name: Resource name (defaults to table_name from config)
        primary_key: Primary key columns (overrides config if provided)
        write_disposition: Write disposition (overrides config if provided)
        columns: Column definitions for the resource
        table_format: Table format for certain destinations
        file_format: File format for filesystem destinations
        schema_contract: Schema contract settings (defaults to evolve mode)
        table_name: Callable to generate dynamic table names
        max_table_nesting: Maximum nesting level for nested data
        selected: Whether this resource is selected for loading
        merge_key: Merge key for merge operations
        parallelized: Whether to parallelize this resource
        **resource_kwargs: Additional keyword arguments for dlt.resource

    Returns:
        dlt.Resource: Configured DLT resource with unified schema support

    Example:
        ```python
        from estat_api_dlt_helper import EstatDltConfig
        from estat_api_dlt_helper.loader.unified_schema_resource import create_unified_estat_resource

        # Configuration with multiple statsDataIds that have different schemas
        config = EstatDltConfig(
            source={
                "statsDataId": ["0004028473", "0004028474", "0004028475"],
                "app_id": "YOUR_APP_ID"
            },
            destination={
                "table_name": "unified_stats",
                "dataset_name": "estat_data"
            }
        )

        # Create resource with unified schema
        resource = create_unified_estat_resource(config)

        # Run in pipeline without schema errors
        import dlt
        pipeline = dlt.pipeline(
            pipeline_name="estat_unified",
            destination="duckdb",
            dataset_name="estat_data"
        )
        pipeline.run(resource)
        ```

    Note:
        This resource is recommended when loading multiple statsDataIds that
        may have different metadata structures (e.g., some have parent_code
        in time_metadata while others don't).
    """

    # Prepare API parameters
    from ..loader.dlt_resource import _create_api_params

    api_params = _create_api_params(config)

    # Get stats data IDs (ensure it's a list)
    stats_data_ids = config.source.statsDataId
    if isinstance(stats_data_ids, str):
        stats_data_ids = [stats_data_ids]

    # Prepare resource configuration
    resource_config: Dict[str, Any] = {
        "name": name or config.destination.table_name,
        "write_disposition": write_disposition or config.destination.write_disposition,
        "schema_contract": schema_contract
        or {
            "tables": "evolve",
            "columns": "evolve",
            "data_type": "freeze",
        },
    }

    # Add primary key for merge disposition
    if primary_key is not None:
        resource_config["primary_key"] = primary_key
    elif (
        config.destination.write_disposition == "merge"
        and config.destination.primary_key
    ):
        pk = config.destination.primary_key
        if isinstance(pk, str):
            pk = [pk]
        resource_config["primary_key"] = pk

    # Add optional resource parameters
    optional_params = {
        "columns": columns,
        "table_format": table_format,
        "file_format": file_format,
        "table_name": table_name,
        "max_table_nesting": max_table_nesting,
        "selected": selected,
        "merge_key": merge_key,
        "parallelized": parallelized,
    }

    for key, value in optional_params.items():
        if value is not None:
            resource_config[key] = value

    # Add any additional resource kwargs
    resource_config.update(resource_kwargs)

    @dlt.resource(**resource_config)
    def unified_estat_data() -> Generator[Dict[str, Any], None, None]:
        """Generator function for unified e-Stat data."""
        client_kwargs: Dict[str, Any] = {"app_id": config.source.app_id}
        if config.timeout is not None:
            client_kwargs["timeout"] = config.timeout
        client = EstatApiClient(**client_kwargs)

        try:
            logger.info(
                f"Processing {len(stats_data_ids)} stats data IDs with unified schema"
            )

            # Process each stats data ID
            for stats_data_id in stats_data_ids:
                yield from _fetch_unified_estat_data(
                    client=client,
                    stats_data_id=stats_data_id,
                    params=api_params,
                    limit=config.source.limit,
                    maximum_offset=config.source.maximum_offset,
                )
        finally:
            client.close()

    return unified_estat_data()

APIリファレンス

メイン設定クラス

EstatDltConfig

SourceConfig

validate_stats_data_id(v) classmethod

DestinationConfig

validate_primary_key(v, info) classmethod

APIクライアント

EstatApiClient

__init__(app_id, base_url=None, timeout=60)

close()

get_stats_data(stats_data_id, start_position=1, limit=100000, meta_get_flg='Y', cnt_get_flg='N', explanation_get_flg='Y', annotation_get_flg='Y', replace_sp_chars='0', lang='J', **additional_params)

get_stats_data_generator(stats_data_id, limit_per_request=100000, **kwargs)

get_stats_list(search_word=None, survey_years=None, stats_code=None, **kwargs)

データ解析

parse_response

データローダー関数

estat_table

estat_table(stats_data_id, app_id=dlt.secrets.value, table_name=None, write_disposition='replace', primary_key=None, incremental=None, limit=_UNSET, maximum_offset=_UNSET, timeout=_UNSET, **api_params)

estat_source

estat_source(stats_data_ids=None, tables=None, app_id=dlt.secrets.value, write_disposition='replace', primary_key=None, incremental=None, limit=100000, maximum_offset=None, timeout=60, **api_params)

load_estat_data

create_estat_resource

create_estat_pipeline

create_estat_source

create_unified_estat_resource

`validate_stats_data_id(v)` `classmethod`

`validate_primary_key(v, info)` `classmethod`

`init(app_id, base_url=None, timeout=60)`

`close()`

`get_stats_data(stats_data_id, start_position=1, limit=100000, meta_get_flg='Y', cnt_get_flg='N', explanation_get_flg='Y', annotation_get_flg='Y', replace_sp_chars='0', lang='J', **additional_params)`

`get_stats_data_generator(stats_data_id, limit_per_request=100000, **kwargs)`

`get_stats_list(search_word=None, survey_years=None, stats_code=None, **kwargs)`

`estat_table(stats_data_id, app_id=dlt.secrets.value, table_name=None, write_disposition='replace', primary_key=None, incremental=None, limit=_UNSET, maximum_offset=_UNSET, timeout=_UNSET, **api_params)`

`estat_source(stats_data_ids=None, tables=None, app_id=dlt.secrets.value, write_disposition='replace', primary_key=None, incremental=None, limit=100000, maximum_offset=None, timeout=60, **api_params)`