tessl/pypi-ipinfo
Official Python library for IPInfo IP address geolocation and data lookups
—
Pending
async-handlers.mddocs/
Asynchronous Handlers
Asynchronous API clients for IP geolocation lookups using aiohttp. Provides concurrent IP lookups with both Core API and Lite API access, advanced timeout controls, and efficient batch processing.
Capabilities
Async Handler Factory Functions
Factory functions that create and return configured asynchronous handler instances.
def getHandlerAsync(access_token=None, **kwargs):
"""
Create and return asynchronous Handler object for Core API access.
Parameters:
- access_token (str, optional): IPinfo API access token
- **kwargs: Additional configuration options
Returns:
AsyncHandler: Configured AsyncHandler instance
"""
def getHandlerAsyncLite(access_token=None, **kwargs):
"""
Create and return asynchronous HandlerLite object for Lite API access.
Parameters:
- access_token (str, optional): IPinfo API access token
- **kwargs: Additional configuration options
Returns:
AsyncHandlerLite: Configured AsyncHandlerLite instance
"""Async Core API Handler
Asynchronous client for the IPinfo Core API supporting concurrent IP lookups, batch operations with configurable concurrency levels, and advanced timeout management.
class AsyncHandler:
def __init__(self, access_token=None, **kwargs):
"""
Initialize AsyncHandler with configuration options.
Parameters:
- access_token (str, optional): IPinfo API access token
- countries (dict, optional): Custom country code to name mappings
- eu_countries (list, optional): Custom list of EU country codes
- countries_flags (dict, optional): Custom country flag mappings
- countries_currencies (dict, optional): Custom country currency mappings
- continent (dict, optional): Custom continent mappings
- request_options (dict, optional): HTTP request configuration
- cache_options (dict, optional): Cache configuration (maxsize, ttl)
- cache (CacheInterface, optional): Custom cache implementation
- headers (dict, optional): Custom HTTP headers
"""
async def init(self):
"""
Initialize internal aiohttp connection pool.
This is optional as the pool is initialized lazily when needed,
but can be used for non-lazy initialization.
"""
async def deinit(self):
"""
Deinitialize the async handler and cleanup resources.
Required for proper resource cleanup in long-running processes.
Must be called to close aiohttp session.
"""
async def getDetails(self, ip_address=None, timeout=None):
"""
Get details for specified IP address asynchronously.
Parameters:
- ip_address (str|IPv4Address|IPv6Address, optional): IP address to lookup
- timeout (int, optional): Request timeout override
Returns:
Details: IP address details object
Raises:
RequestQuotaExceededError: When API quota exceeded
APIError: When API returns error response
"""
async def getBatchDetails(self, ip_addresses, batch_size=None, timeout_per_batch=5, timeout_total=None, raise_on_fail=True):
"""
Get details for multiple IP addresses with concurrent batch processing.
Parameters:
- ip_addresses (list): List of IP addresses to lookup
- batch_size (int, optional): Batch size (default: 1000, max: 1000)
- timeout_per_batch (int, optional): Timeout per batch request (default: 5)
- timeout_total (int, optional): Total operation timeout
- raise_on_fail (bool, optional): Whether to raise on errors (default: True)
Returns:
dict: Mapping of IP addresses to Details objects or error info
Raises:
RequestQuotaExceededError: When API quota exceeded
TimeoutExceededError: When timeout limits exceeded
"""
async def getBatchDetailsIter(self, ip_addresses, batch_size=None, raise_on_fail=True):
"""
Async iterator for batch details lookup.
Parameters:
- ip_addresses (list): List of IP addresses to lookup
- batch_size (int, optional): Batch size (default: 1000)
- raise_on_fail (bool, optional): Whether to raise on errors (default: True)
Yields:
tuple: (ip_address, details) pairs
Raises:
RequestQuotaExceededError: When API quota exceeded
"""Async Lite API Handler
Asynchronous client for the IPinfo Lite API providing fast, concurrent IP geolocation lookups with simplified response data.
class AsyncHandlerLite:
def __init__(self, access_token=None, **kwargs):
"""
Initialize AsyncHandlerLite with configuration options.
Parameters: Same as AsyncHandler class
"""
async def init(self):
"""
Initialize internal aiohttp connection pool.
Optional lazy initialization for connection pool.
"""
async def deinit(self):
"""
Deinitialize the async handler and cleanup resources.
Required for proper resource cleanup.
"""
async def getDetails(self, ip_address=None, timeout=None):
"""
Get details for specified IP address using Lite API asynchronously.
Parameters:
- ip_address (str|IPv4Address|IPv6Address, optional): IP address to lookup
- timeout (int, optional): Request timeout override
Returns:
Details: IP address details object with Lite API fields
Raises:
RequestQuotaExceededError: When API quota exceeded
APIError: When API returns error response
"""Usage Examples
Basic Async Usage
import asyncio
import ipinfo
async def main():
# Create async handler
handler = ipinfo.getHandlerAsync('your_access_token')
try:
# Single IP lookup
details = await handler.getDetails('8.8.8.8')
print(f"Location: {details.city}, {details.region}, {details.country}")
print(f"ISP: {details.org}")
finally:
# Always cleanup resources
await handler.deinit()
# Run the async function
asyncio.run(main())Concurrent Lookups
import asyncio
import ipinfo
async def lookup_ip(handler, ip):
"""Lookup single IP with error handling"""
try:
details = await handler.getDetails(ip)
return ip, details.city, details.country
except Exception as e:
return ip, None, str(e)
async def main():
handler = ipinfo.getHandlerAsync('your_access_token')
try:
# Concurrent individual lookups
ips = ['8.8.8.8', '1.1.1.1', '208.67.222.222']
tasks = [lookup_ip(handler, ip) for ip in ips]
results = await asyncio.gather(*tasks)
for ip, city, country in results:
print(f"{ip}: {city}, {country}")
finally:
await handler.deinit()
asyncio.run(main())Batch Processing
import asyncio
import ipinfo
async def main():
handler = ipinfo.getHandlerAsync('your_access_token')
try:
# Efficient batch processing
ips = ['8.8.8.8', '1.1.1.1', '208.67.222.222'] * 100
results = await handler.getBatchDetails(
ips,
batch_size=50,
timeout_per_batch=10,
timeout_total=60
)
print(f"Processed {len(results)} IPs")
for ip, details in results.items():
if hasattr(details, 'city'):
print(f"{ip}: {details.city}")
finally:
await handler.deinit()
asyncio.run(main())Context Manager Pattern
import asyncio
import ipinfo
from contextlib import asynccontextmanager
@asynccontextmanager
async def ipinfo_handler(access_token):
"""Context manager for proper resource management"""
handler = ipinfo.getHandlerAsync(access_token)
await handler.init()
try:
yield handler
finally:
await handler.deinit()
async def main():
async with ipinfo_handler('your_access_token') as handler:
details = await handler.getDetails('8.8.8.8')
print(f"City: {details.city}")
asyncio.run(main())Async Iterator Usage
import asyncio
import ipinfo
async def main():
handler = ipinfo.getHandlerAsync('your_access_token')
try:
ips = ['8.8.8.8', '1.1.1.1', '208.67.222.222'] * 10
async for ip, details in handler.getBatchDetailsIter(ips, batch_size=10):
if hasattr(details, 'city'):
print(f"{ip}: {details.city}, {details.country}")
finally:
await handler.deinit()
asyncio.run(main())Lite API Async Usage
import asyncio
import ipinfo
async def main():
handler = ipinfo.getHandlerAsyncLite('your_access_token')
try:
details = await handler.getDetails('8.8.8.8')
print(f"Country: {details.country}")
print(f"Country Code: {details.country_code}")
finally:
await handler.deinit()
asyncio.run(main())Error Handling
import asyncio
import ipinfo
from ipinfo.exceptions import RequestQuotaExceededError, TimeoutExceededError
from ipinfo.error import APIError
async def main():
handler = ipinfo.getHandlerAsync('your_access_token')
try:
details = await handler.getDetails('8.8.8.8')
print(details.city)
except RequestQuotaExceededError:
print("Monthly quota exceeded")
except TimeoutExceededError:
print("Request timed out")
except APIError as e:
print(f"API error {e.error_code}: {e.error_json}")
except Exception as e:
print(f"Unexpected error: {e}")
finally:
await handler.deinit()
asyncio.run(main())Types
# Async handler configuration (same as sync handlers)
AsyncHandlerOptions = {
'countries': dict,
'eu_countries': list,
'countries_flags': dict,
'countries_currencies': dict,
'continent': dict,
'request_options': dict,
'cache_options': dict,
'cache': CacheInterface,
'headers': dict
}
# Async batch result
AsyncBatchResult = dict # IP address -> Details object or error info
# Async iterator result
AsyncIterResult = tuple # (ip_address, details)Install with Tessl CLI
npx tessl i tessl/pypi-ipinfo