Package `python_odpt`

python-odpt

Python client for ODPT(Open Data of Public Transportation) API This package is a Python client for the ODPT (Open Data of Public Transportation) API. This library is automatically generated from the ODPT OpenAPI specifications, providing a fully typed and easy-to-use interface to interact with Japan's public transportation data.

OpenAPI spec: https://github.com/sophie-app/odpt-openapi
Generated by: https://github.com/openapi-generators/openapi-python-client

Thanks for great works!

Docs PyPI

Installation

You can install this package with either of the following commands:

# pip
pip install python-odpt

# poetry
poetry add python-odpt

Usage

Getting an ODPT Access Token

First, an ODPT access token is required to authenticate your API requests. Access https://developer.odpt.org/ and submit the user registration form. After your registration is accepted, you can log in to the developer site and obtain your access token from the top right menu.

Get started

To use the library, you need to import the core client and the specific API functions you intend to use. The API endpoints are organized into separate modules, which keeps the namespace clean. To connect to the ODPT API, you need to initialize an Client. This client handles the base URL and connection pooling for your requests.

# Import the authenticated client
from python_odpt import Client

# Import specific API endpoints (Example: searching something from datas)
from python_odpt.api import data_search

API_URL = "https://api.odpt.org/api/v4/"
ACCESS_TOKEN = "YOUR_ACCESS_TOKEN"

# Create the client instance
client = Client(API_URL)

res=data_search.sync_detailed(client=client,
    aclconsumer_key=ACCESS_TOKEN, rdf_type="odpt:BusstopPole", 
    predicate={"odpt:operator":"odpt.Operator:Toei"})

print(res.status_code)
print(res.parsed[0])

[!WARNING] Do NOT write your access token on your code. This is just for a example, the access token sould be provided by other safety way. DO NOT PUBLISH CODE INCLUDING YOUR ACCESS TOKEN.

First API Request Example

The python-odpt library supports both synchronous and asynchronous requests. The functions typically end in .sync (for synchronous execution) or .asyncio (for asynchronous execution using async/await). Below is an example of fetching train operation information synchronously:

from python_odpt.models import OdptTrainInformation
from python_odpt.api import get_train_information

# Fetch train information synchronously
# You can pass query parameters as keyword arguments
train_info_list = get_train_information.sync(
    client=client,
    aclconsumer_key=ACCESS_TOKEN,
    # Example query parameters:
    # odpt_operator="odpt.Operator:JR-East"
)

if train_info_list:
    print(f"Retrieved {len(train_info_list)} records.")
else:
    print("No data retrieved.")

If you are writing an asynchronous application (e.g., using FastAPI, Discord.py, or standard asyncio), you can use the asynchronous variant:

import asyncio

async def fetch_data():
    train_info_list = await get_train_information.asyncio(client=client,
                                   aclconsumer_key=ACCESS_TOKEN)
    print(f"Retrieved asynchronously: {len(train_info_list)}")

asyncio.run(fetch_data())

How to Inspect and Use Returned Data

The ODPT API responses are automatically parsed into Python dataclasses/models by the library. This gives you full IDE autocompletion and type safety, eliminating the need to guess dictionary keys.

Accessing Attributes

You can access properties directly using dot notation. Attributes that correspond to Japanese localizations or nested objects are typically wrapped in child classes.

for train_info in train_info_list:
    # Access basic string properties
    print(f"Railway ID: {train_info.odptrailway}")

    # Access nested/localized properties (e.g., Multilingual text)
    if train_info.odpttrain_information_text:
        print(f"Information (JA): {train_info.odpttrain_information_text.ja}")
        print(f"Information (EN): {train_info.odpttrain_information_text.en}")

    # Access date-time objects (automatically parsed into datetime instances)
    print(f"Time of generation: {train_info.dcdate}")
    print("-" * 20)

Fallback to Dictionary or JSON

If you need to serialize the data for an API response, or if you prefer working with raw Python dictionaries, you can easily convert the model back:

import json

# Convert a single model to a dictionary
raw_dict = train_info_list[0].to_dict()
print(raw_dict["odpt:railway"])

# Or convert to JSON
json_string = json.dumps(raw_dict, ensure_ascii=False, indent=2)
print(json_string)

Handling Detailed Responses

If you need access to the raw HTTP response (for example, to inspect headers or HTTP status codes), you can use .sync_detailed or .asyncio_detailed instead. This will return a Response object instead of just the parsed model.

response_obj = get_odpt_train_information.sync_detailed(client=client, aclconsumer_key=ACCESS_TOKEN)

print(f"Status Code: {response_obj.status_code}")
print(f"Headers: {response_obj.headers}")

# The parsed data is available in the 'parsed' attribute
if response_obj.parsed:
    for info in response_obj.parsed:
        print(info.odptrailway)

ChangeLog

v0.1.0 First release
v0.1.2 Add missing dependencies
v0.1.3 Fix handling for nullable field
v0.1.4 Fix datadump API typing
v0.1.5 Fix wrong endpoint
v0.1.6 Change import endpoint and update README

# Before - Redundant
from python_odpt.api.default import data_search_operations_search
# After - Simple
from python_odpt.api import data_search

The older endpoints will be held for compatibility. These will be removed on the next minor version up.

A client library for accessing ODPT API

Sub-modules

python_odpt.api
python_odpt.client
python_odpt.errors: Contains shared errors types that can be raised from API functions
python_odpt.models: Contains all the data models used in inputs/outputs
python_odpt.types: Contains some shared types for properties

Classes

class AuthenticatedClient (base_url: str, token: str, prefix: str = 'Bearer', auth_header_name: str = 'Authorization', *, raise_on_unexpected_status: bool = False, cookies: Dict[str, str] = _Nothing.NOTHING, headers: Dict[str, str] = _Nothing.NOTHING, timeout: Optional[httpx.Timeout] = None, verify_ssl: Union[str, bool, ssl.SSLContext] = True, follow_redirects: bool = False, httpx_args: Dict[str, Any] = _Nothing.NOTHING)

A Client which has been authenticated for use on secured endpoints

The following are accepted as keyword arguments and will be used to construct httpx Clients internally:

<code>base\_url</code>: The base URL for the API, all requests are made to a relative path to this URL

<code>cookies</code>: A dictionary of cookies to be sent with every request

<code>headers</code>: A dictionary of headers to be sent with every request

<code>timeout</code>: The maximum amount of a time a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.

<code>verify\_ssl</code>: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.

<code>follow\_redirects</code>: Whether or not to follow redirects. Default value is False.

<code>httpx\_args</code>: A dictionary of additional arguments to be passed to the <code>httpx.Client</code> and <code>httpx.AsyncClient</code> constructor.

Attributes

raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a status code that was not documented in the source OpenAPI document. Can also be provided as a keyword argument to the constructor.
token: The token to use for authentication
prefix: The prefix to use for the Authorization header
auth_header_name: The name of the Authorization header

Method generated by attrs for class AuthenticatedClient.

Expand source code

class AuthenticatedClient:
    """A Client which has been authenticated for use on secured endpoints

    The following are accepted as keyword arguments and will be used to construct httpx Clients internally:

        ``base_url``: The base URL for the API, all requests are made to a relative path to this URL

        ``cookies``: A dictionary of cookies to be sent with every request

        ``headers``: A dictionary of headers to be sent with every request

        ``timeout``: The maximum amount of a time a request can take. API functions will raise
        httpx.TimeoutException if this is exceeded.

        ``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
        but can be set to False for testing purposes.

        ``follow_redirects``: Whether or not to follow redirects. Default value is False.

        ``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.


    Attributes:
        raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
            status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
            argument to the constructor.
        token: The token to use for authentication
        prefix: The prefix to use for the Authorization header
        auth_header_name: The name of the Authorization header
    """

    raise_on_unexpected_status: bool = field(default=False, kw_only=True)
    _base_url: str = field(alias="base_url")
    _cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies")
    _headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers")
    _timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout")
    _verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl")
    _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects")
    _httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args")
    _client: Optional[httpx.Client] = field(default=None, init=False)
    _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False)

    token: str
    prefix: str = "Bearer"
    auth_header_name: str = "Authorization"

    def with_headers(self, headers: Dict[str, str]) -> "AuthenticatedClient":
        """Get a new client matching this one with additional headers"""
        if self._client is not None:
            self._client.headers.update(headers)
        if self._async_client is not None:
            self._async_client.headers.update(headers)
        return evolve(self, headers={**self._headers, **headers})

    def with_cookies(self, cookies: Dict[str, str]) -> "AuthenticatedClient":
        """Get a new client matching this one with additional cookies"""
        if self._client is not None:
            self._client.cookies.update(cookies)
        if self._async_client is not None:
            self._async_client.cookies.update(cookies)
        return evolve(self, cookies={**self._cookies, **cookies})

    def with_timeout(self, timeout: httpx.Timeout) -> "AuthenticatedClient":
        """Get a new client matching this one with a new timeout (in seconds)"""
        if self._client is not None:
            self._client.timeout = timeout
        if self._async_client is not None:
            self._async_client.timeout = timeout
        return evolve(self, timeout=timeout)

    def set_httpx_client(self, client: httpx.Client) -> "AuthenticatedClient":
        """Manually set the underlying httpx.Client

        **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
        """
        self._client = client
        return self

    def get_httpx_client(self) -> httpx.Client:
        """Get the underlying httpx.Client, constructing a new one if not previously set"""
        if self._client is None:
            self._headers[self.auth_header_name] = f"{self.prefix} {self.token}" if self.prefix else self.token
            self._client = httpx.Client(
                base_url=self._base_url,
                cookies=self._cookies,
                headers=self._headers,
                timeout=self._timeout,
                verify=self._verify_ssl,
                follow_redirects=self._follow_redirects,
                **self._httpx_args,
            )
        return self._client

    def __enter__(self) -> "AuthenticatedClient":
        """Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
        self.get_httpx_client().__enter__()
        return self

    def __exit__(self, *args: Any, **kwargs: Any) -> None:
        """Exit a context manager for internal httpx.Client (see httpx docs)"""
        self.get_httpx_client().__exit__(*args, **kwargs)

    def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "AuthenticatedClient":
        """Manually the underlying httpx.AsyncClient

        **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
        """
        self._async_client = async_client
        return self

    def get_async_httpx_client(self) -> httpx.AsyncClient:
        """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
        if self._async_client is None:
            self._headers[self.auth_header_name] = f"{self.prefix} {self.token}" if self.prefix else self.token
            self._async_client = httpx.AsyncClient(
                base_url=self._base_url,
                cookies=self._cookies,
                headers=self._headers,
                timeout=self._timeout,
                verify=self._verify_ssl,
                follow_redirects=self._follow_redirects,
                **self._httpx_args,
            )
        return self._async_client

    async def __aenter__(self) -> "AuthenticatedClient":
        """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
        await self.get_async_httpx_client().__aenter__()
        return self

    async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
        """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
        await self.get_async_httpx_client().__aexit__(*args, **kwargs)

Instance variables

var auth_header_name : str
var prefix : str
var raise_on_unexpected_status : bool
var token : str

Methods

def get_async_httpx_client(self) ‑> httpx.AsyncClient: Get the underlying httpx.AsyncClient, constructing a new one if not previously set
def get_httpx_client(self) ‑> httpx.Client: Get the underlying httpx.Client, constructing a new one if not previously set
def set_async_httpx_client(self, async_client: httpx.AsyncClient) ‑> AuthenticatedClient: Manually the underlying httpx.AsyncClient

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.
def set_httpx_client(self, client: httpx.Client) ‑> AuthenticatedClient: Manually set the underlying httpx.Client

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.
def with_cookies(self, cookies: Dict[str, str]) ‑> AuthenticatedClient: Get a new client matching this one with additional cookies
def with_headers(self, headers: Dict[str, str]) ‑> AuthenticatedClient: Get a new client matching this one with additional headers
def with_timeout(self, timeout: httpx.Timeout) ‑> AuthenticatedClient: Get a new client matching this one with a new timeout (in seconds)

class Client (base_url: str, *, raise_on_unexpected_status: bool = False, cookies: Dict[str, str] = _Nothing.NOTHING, headers: Dict[str, str] = _Nothing.NOTHING, timeout: Optional[httpx.Timeout] = None, verify_ssl: Union[str, bool, ssl.SSLContext] = True, follow_redirects: bool = False, httpx_args: Dict[str, Any] = _Nothing.NOTHING)

A class for keeping track of data related to the API

The following are accepted as keyword arguments and will be used to construct httpx Clients internally:

<code>base\_url</code>: The base URL for the API, all requests are made to a relative path to this URL

<code>cookies</code>: A dictionary of cookies to be sent with every request

<code>headers</code>: A dictionary of headers to be sent with every request

<code>timeout</code>: The maximum amount of a time a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.

<code>verify\_ssl</code>: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.

<code>follow\_redirects</code>: Whether or not to follow redirects. Default value is False.

<code>httpx\_args</code>: A dictionary of additional arguments to be passed to the <code>httpx.Client</code> and <code>httpx.AsyncClient</code> constructor.

Attributes

raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a status code that was not documented in the source OpenAPI document. Can also be provided as a keyword argument to the constructor.

Method generated by attrs for class Client.

Expand source code

class Client:
    """A class for keeping track of data related to the API

    The following are accepted as keyword arguments and will be used to construct httpx Clients internally:

        ``base_url``: The base URL for the API, all requests are made to a relative path to this URL

        ``cookies``: A dictionary of cookies to be sent with every request

        ``headers``: A dictionary of headers to be sent with every request

        ``timeout``: The maximum amount of a time a request can take. API functions will raise
        httpx.TimeoutException if this is exceeded.

        ``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
        but can be set to False for testing purposes.

        ``follow_redirects``: Whether or not to follow redirects. Default value is False.

        ``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.


    Attributes:
        raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
            status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
            argument to the constructor.
    """

    raise_on_unexpected_status: bool = field(default=False, kw_only=True)
    _base_url: str = field(alias="base_url")
    _cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies")
    _headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers")
    _timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout")
    _verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl")
    _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects")
    _httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args")
    _client: Optional[httpx.Client] = field(default=None, init=False)
    _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False)

    def with_headers(self, headers: Dict[str, str]) -> "Client":
        """Get a new client matching this one with additional headers"""
        if self._client is not None:
            self._client.headers.update(headers)
        if self._async_client is not None:
            self._async_client.headers.update(headers)
        return evolve(self, headers={**self._headers, **headers})

    def with_cookies(self, cookies: Dict[str, str]) -> "Client":
        """Get a new client matching this one with additional cookies"""
        if self._client is not None:
            self._client.cookies.update(cookies)
        if self._async_client is not None:
            self._async_client.cookies.update(cookies)
        return evolve(self, cookies={**self._cookies, **cookies})

    def with_timeout(self, timeout: httpx.Timeout) -> "Client":
        """Get a new client matching this one with a new timeout (in seconds)"""
        if self._client is not None:
            self._client.timeout = timeout
        if self._async_client is not None:
            self._async_client.timeout = timeout
        return evolve(self, timeout=timeout)

    def set_httpx_client(self, client: httpx.Client) -> "Client":
        """Manually set the underlying httpx.Client

        **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
        """
        self._client = client
        return self

    def get_httpx_client(self) -> httpx.Client:
        """Get the underlying httpx.Client, constructing a new one if not previously set"""
        if self._client is None:
            self._client = httpx.Client(
                base_url=self._base_url,
                cookies=self._cookies,
                headers=self._headers,
                timeout=self._timeout,
                verify=self._verify_ssl,
                follow_redirects=self._follow_redirects,
                **self._httpx_args,
            )
        return self._client

    def __enter__(self) -> "Client":
        """Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
        self.get_httpx_client().__enter__()
        return self

    def __exit__(self, *args: Any, **kwargs: Any) -> None:
        """Exit a context manager for internal httpx.Client (see httpx docs)"""
        self.get_httpx_client().__exit__(*args, **kwargs)

    def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "Client":
        """Manually the underlying httpx.AsyncClient

        **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
        """
        self._async_client = async_client
        return self

    def get_async_httpx_client(self) -> httpx.AsyncClient:
        """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
        if self._async_client is None:
            self._async_client = httpx.AsyncClient(
                base_url=self._base_url,
                cookies=self._cookies,
                headers=self._headers,
                timeout=self._timeout,
                verify=self._verify_ssl,
                follow_redirects=self._follow_redirects,
                **self._httpx_args,
            )
        return self._async_client

    async def __aenter__(self) -> "Client":
        """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
        await self.get_async_httpx_client().__aenter__()
        return self

    async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
        """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
        await self.get_async_httpx_client().__aexit__(*args, **kwargs)

Instance variables

var raise_on_unexpected_status : bool

Methods

def get_async_httpx_client(self) ‑> httpx.AsyncClient: Get the underlying httpx.AsyncClient, constructing a new one if not previously set
def get_httpx_client(self) ‑> httpx.Client: Get the underlying httpx.Client, constructing a new one if not previously set
def set_async_httpx_client(self, async_client: httpx.AsyncClient) ‑> Client: Manually the underlying httpx.AsyncClient

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.
def set_httpx_client(self, client: httpx.Client) ‑> Client: Manually set the underlying httpx.Client

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.
def with_cookies(self, cookies: Dict[str, str]) ‑> Client: Get a new client matching this one with additional cookies
def with_headers(self, headers: Dict[str, str]) ‑> Client: Get a new client matching this one with additional headers
def with_timeout(self, timeout: httpx.Timeout) ‑> Client: Get a new client matching this one with a new timeout (in seconds)