Python Examples

import requests
import json
from typing import Dict, List, Optional

class IPOAlertsClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.ipoalerts.in"
        self.session = requests.Session()
        self.session.headers.update({
            "x-api-key": api_key,
            "Content-Type": "application/json"
        })
    
    def get_ipos(self, status: Optional[str] = None, 
                 type: Optional[str] = None,
                 page: int = 1,
                 limit: int = 10) -> Dict:
        """Get IPOs with optional filtering and pagination."""
        params = {
            "page": page,
            "limit": limit
        }
        
        if status:
            params["status"] = status
        if type:
            params["type"] = type
        
        response = self.session.get(f"{self.base_url}/ipos", params=params)
        response.raise_for_status()
        return response.json()
    
    def get_ipo(self, identifier: str) -> Dict:
        """Get a specific IPO by ID, symbol, or slug."""
        response = self.session.get(f"{self.base_url}/ipos/{identifier}")
        response.raise_for_status()
        return response.json()

# Usage
client = IPOAlertsClient("YOUR_API_KEY")

# Get upcoming IPOs
upcoming_ipos = client.get_ipos(status="upcoming", type="EQ")
print(f"Found {upcoming_ipos['meta']['count']} upcoming equity IPOs")

for ipo in upcoming_ipos["ipos"]:
    print(f"{ipo['name']} ({ipo['symbol']}) - {ipo['priceRange']}")

# Get specific IPO
ipo_details = client.get_ipo("example-company-limited")
print(f"IPO Details: {ipo_details['ipo']['name']}")

import time
import hashlib
from functools import wraps
from typing import Dict, Any

class RateLimitedIPOAlertsClient(IPOAlertsClient):
    def __init__(self, api_key: str, requests_per_minute: int = 6):
        super().__init__(api_key)
        self.requests_per_minute = requests_per_minute
        self.request_times = []
        self.cache = {}
        self.cache_duration = 300  # 5 minutes
    
    def _wait_if_needed(self):
        """Wait if we're approaching the rate limit."""
        now = time.time()
        # Remove requests older than 1 minute
        self.request_times = [t for t in self.request_times if now - t < 60]
        
        if len(self.request_times) >= self.requests_per_minute:
            sleep_time = 60 - (now - self.request_times[0])
            if sleep_time > 0:
                time.sleep(sleep_time)
    
    def _get_cache_key(self, endpoint: str, params: Dict) -> str:
        """Generate cache key for request."""
        key_data = f"{endpoint}:{json.dumps(params, sort_keys=True)}"
        return hashlib.md5(key_data.encode()).hexdigest()
    
    def _is_cache_valid(self, cache_key: str) -> bool:
        """Check if cached data is still valid."""
        if cache_key not in self.cache:
            return False
        
        cached_time, _ = self.cache[cache_key]
        return time.time() - cached_time < self.cache_duration
    
    def get_ipos(self, **kwargs) -> Dict:
        """Get IPOs with caching and rate limiting."""
        cache_key = self._get_cache_key("/ipos", kwargs)
        
        if self._is_cache_valid(cache_key):
            print("Returning cached data")
            return self.cache[cache_key][1]
        
        self._wait_if_needed()
        result = super().get_ipos(**kwargs)
        
        # Cache the result
        self.cache[cache_key] = (time.time(), result)
        self.request_times.append(time.time())
        
        return result

# Usage
client = RateLimitedIPOAlertsClient("YOUR_API_KEY")

# This will be cached for 5 minutes
ipos1 = client.get_ipos(status="upcoming")
ipos2 = client.get_ipos(status="upcoming")  # Returns cached data