Skip to content

Client Reference

tickertape.client.TickerTapeClient

Asynchronous client for interacting with TickerTape.

Example:
    async with TickerTapeClient() as client:
        # Reads from persistent cache (.cache/tickertape) or fetches live XML if not cached:
        mf_sitemap = await client.sitemap.get("mf")

        # Force live refresh and update disk cache:
        refreshed = await client.sitemap.refresh("mf")

        # Fetch fund details and ISIN:
        fund = await client.mf.get("quant-infrastructure-fund-M_QUNG")
        print(fund.isin, fund.nav)
Source code in src/tickertape/client.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class TickerTapeClient:
    """Asynchronous client for interacting with TickerTape.

    ```
    Example:
        async with TickerTapeClient() as client:
            # Reads from persistent cache (.cache/tickertape) or fetches live XML if not cached:
            mf_sitemap = await client.sitemap.get("mf")

            # Force live refresh and update disk cache:
            refreshed = await client.sitemap.refresh("mf")

            # Fetch fund details and ISIN:
            fund = await client.mf.get("quant-infrastructure-fund-M_QUNG")
            print(fund.isin, fund.nav)
    ```
    """

    def __init__(
        self,
        base_url: str = BASE_URL,
        timeout: float = 30.0,
        cache_dir: Path | str = DEFAULT_CACHE_DIR,
        http_client: Optional[httpx.AsyncClient] = None,
        headers: Optional[dict[str, str]] = None,
        sitemap_urls: Optional[dict[str, str]] = None,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.cache_dir = Path(cache_dir)
        self.sitemap_urls = dict(sitemap_urls or SITEMAP_URLS)

        req_headers = dict(DEFAULT_HEADERS)
        if headers:
            req_headers.update(headers)
        self._default_headers = req_headers

        self._owns_http_client = http_client is None
        self._http_client = http_client or httpx.AsyncClient(
            headers=self._default_headers,
            timeout=self.timeout,
            follow_redirects=True,
        )

        # Storage & persistent cache manager
        self.cache_manager = SitemapCacheManager(cache_dir=self.cache_dir)

        # Resource sub-clients
        self.sitemap = SitemapResource(self)
        self.mf = MutualFundsResource(self)

    async def __aenter__(self) -> TickerTapeClient:
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        await self.close()

    async def close(self) -> None:
        """Close the underlying HTTP transport."""
        if self._owns_http_client and not self._http_client.is_closed:
            await self._http_client.aclose()

    def _build_url(self, url_or_path: str) -> str:
        """Build full URL if given relative path."""
        url = url_or_path.strip()
        if url.startswith(("http://", "https://")):
            return url
        if not url.startswith("/"):
            url = f"/{url}"
        return f"{self.base_url}{url}"

    async def request(
        self,
        method: str,
        url_or_path: str,
        *,
        params: Optional[dict[str, Any]] = None,
        headers: Optional[dict[str, str]] = None,
        **kwargs: Any,
    ) -> str:
        """Execute an HTTP request and return text response.

        Args:
            method: HTTP method ('GET', 'POST', etc.)
            url_or_path: Full URL or path relative to base_url.
            params: Optional query parameters.
            headers: Optional extra headers.

        Returns:
            Response body string.

        Raises:
            TickerTapeNotFoundError: On HTTP 404.
            TickerTapeHTTPError: On other HTTP 4xx/5xx status codes.
        """
        full_url = self._build_url(url_or_path)

        req_headers = dict(self._default_headers)
        if headers:
            req_headers.update(headers)

        try:
            response = await self._http_client.request(
                method=method,
                url=full_url,
                params=params,
                headers=req_headers,
                **kwargs,
            )
        except httpx.RequestError as exc:
            raise TickerTapeHTTPError(
                f"Request to {full_url} failed: {exc}",
                status_code=0,
            ) from exc

        if response.status_code == 404:
            raise TickerTapeNotFoundError(
                f"Resource not found at {full_url}",
                status_code=404,
                response_data=response.text[:200],
            )

        if response.is_error:
            raise TickerTapeHTTPError(
                f"HTTP {response.status_code} error from {full_url}",
                status_code=response.status_code,
                response_data=response.text[:500],
            )

        return response.text

close async 

close() -> None

Close the underlying HTTP transport.

Source code in src/tickertape/client.py 
90
91
92
93
async def close(self) -> None:
    """Close the underlying HTTP transport."""
    if self._owns_http_client and not self._http_client.is_closed:
        await self._http_client.aclose()

request async 

request(
    method: str,
    url_or_path: str,
    *,
    params: Optional[dict[str, Any]] = None,
    headers: Optional[dict[str, str]] = None,
    **kwargs: Any
) -> str

Execute an HTTP request and return text response.

Parameters:

Name Type Description Default
method str

HTTP method ('GET', 'POST', etc.)

 required
url_or_path str

Full URL or path relative to base_url.

 required
params Optional[dict[str, Any]]

Optional query parameters.

 None
headers Optional[dict[str, str]]

Optional extra headers.

 None

Returns:

Type Description
str

Response body string.

Raises:

Type Description
TickerTapeNotFoundError

On HTTP 404.
TickerTapeHTTPError

On other HTTP 4xx/5xx status codes.
Source code in src/tickertape/client.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
async def request(
    self,
    method: str,
    url_or_path: str,
    *,
    params: Optional[dict[str, Any]] = None,
    headers: Optional[dict[str, str]] = None,
    **kwargs: Any,
) -> str:
    """Execute an HTTP request and return text response.

    Args:
        method: HTTP method ('GET', 'POST', etc.)
        url_or_path: Full URL or path relative to base_url.
        params: Optional query parameters.
        headers: Optional extra headers.

    Returns:
        Response body string.

    Raises:
        TickerTapeNotFoundError: On HTTP 404.
        TickerTapeHTTPError: On other HTTP 4xx/5xx status codes.
    """
    full_url = self._build_url(url_or_path)

    req_headers = dict(self._default_headers)
    if headers:
        req_headers.update(headers)

    try:
        response = await self._http_client.request(
            method=method,
            url=full_url,
            params=params,
            headers=req_headers,
            **kwargs,
        )
    except httpx.RequestError as exc:
        raise TickerTapeHTTPError(
            f"Request to {full_url} failed: {exc}",
            status_code=0,
        ) from exc

    if response.status_code == 404:
        raise TickerTapeNotFoundError(
            f"Resource not found at {full_url}",
            status_code=404,
            response_data=response.text[:200],
        )

    if response.is_error:
        raise TickerTapeHTTPError(
            f"HTTP {response.status_code} error from {full_url}",
            status_code=response.status_code,
            response_data=response.text[:500],
        )

    return response.text