CVR API Implementation Guide¶

import requests
import json
import time
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
import logging
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

@dataclass
class CVRConfig:
    username: str = field(default_factory=lambda: os.getenv('CVR_USERNAME', ''))
    password: str = field(default_factory=lambda: os.getenv('CVR_PASSWORD', ''))
    base_url: str = "http://distribution.virk.dk/cvr-permanent"
    timeout: int = 30
    max_retries: int = 3
    backoff_factor: float = 0.3
    rate_limit_delay: float = 0.2

class CVRError(Exception):
    """Base exception for CVR API errors"""
    pass

class CVRAuthenticationError(CVRError):
    """Authentication failed"""
    pass

class CVRResultWindowError(CVRError):
    """Result window too large (>3000 records)"""
    pass

class CVRTimeoutError(CVRError):
    """Request timed out"""
    pass

class CVRClient:
    def __init__(self, config: CVRConfig = None):
        self.config = config or CVRConfig()
        self.session = self._create_session()
        self.logger = logging.getLogger(__name__)

    def _create_session(self) -> requests.Session:
        """Create HTTP session with retries and connection pooling"""
        session = requests.Session()
        session.auth = (self.config.username, self.config.password)

        # Configure retries
        retry_strategy = Retry(
            total=self.config.max_retries,
            backoff_factor=self.config.backoff_factor,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["POST"]
        )

        adapter = HTTPAdapter(max_retries=retry_strategy, pool_maxsize=10)
        session.mount("http://", adapter)
        session.mount("https://", adapter)

        return session

    def _make_request(self, endpoint: str, query: Dict[str, Any]) -> Dict[str, Any]:
        """Make API request with error handling"""
        url = f"{self.config.base_url}/{endpoint}/_search"
        headers = {'Content-Type': 'application/json'}

        try:
            # Rate limiting
            time.sleep(self.config.rate_limit_delay)

            response = self.session.post(
                url, 
                json=query, 
                headers=headers, 
                timeout=self.config.timeout
            )

            # Handle authentication errors
            if response.status_code == 401:
                raise CVRAuthenticationError("Invalid credentials")

            # Handle result window errors
            if response.status_code == 500:
                error_data = response.json()
                if "Result window is too large" in str(error_data):
                    raise CVRResultWindowError("Query exceeds 3000 result limit. Use scroll API.")
                raise CVRError(f"Server error: {error_data}")

            response.raise_for_status()
            return response.json()

        except requests.exceptions.Timeout:
            raise CVRTimeoutError("Request timed out")
        except requests.exceptions.ConnectionError as e:
            raise CVRError(f"Connection error: {e}")
        except requests.exceptions.RequestException as e:
            raise CVRError(f"Request error: {e}")

    def search_companies(self, query: Dict[str, Any]) -> Dict[str, Any]:
        """Search companies (virksomhed) index"""
        return self._make_request("virksomhed", query)

    def search_participants(self, query: Dict[str, Any]) -> Dict[str, Any]:
        """Search participants (deltager) index"""
        return self._make_request("deltager", query)

    def search_production_units(self, query: Dict[str, Any]) -> Dict[str, Any]:
        """Search production units (produktionsenhed) index"""
        return self._make_request("produktionsenhed", query)

    def get_company_by_cvr(self, cvr_number: str) -> Optional[Dict[str, Any]]:
        """Get company by CVR number"""
        query = {
            "query": {
                "term": {
                    "Vrvirksomhed.cvrNummer": str(cvr_number)
                }
            }
        }

        result = self.search_companies(query)
        hits = result.get("hits", {}).get("hits", [])
        return hits[0]["_source"] if hits else None

    def search_companies_by_name(self, name: str, size: int = 10) -> List[Dict[str, Any]]:
        """Search companies by name with fuzzy matching"""
        query = {
            "query": {
                "bool": {
                    "should": [
                        {
                            "match": {
                                "Vrvirksomhed.navne.navn": {
                                    "query": name,
                                    "fuzziness": "AUTO"
                                }
                            }
                        },
                        {
                            "wildcard": {
                                "Vrvirksomhed.navne.navn": f"*{name.upper()}*"
                            }
                        }
                    ]
                }
            },
            "_source": [
                "Vrvirksomhed.cvrNummer", 
                "Vrvirksomhed.navne", 
                "Vrvirksomhed.virksomhedsstatus"
            ],
            "size": size
        }

        result = self.search_companies(query)
        return [hit["_source"] for hit in result.get("hits", {}).get("hits", [])]

    def get_active_companies_by_industry(self, industry_code: str, size: int = 100) -> List[Dict[str, Any]]:
        """Get active companies by industry code"""
        query = {
            "query": {
                "bool": {
                    "must": [
                        {
                            "term": {
                                "Vrvirksomhed.hovedbranche.branchekode": industry_code
                            }
                        },
                        {
                            "term": {
                                "Vrvirksomhed.virksomhedsstatus.status": "NORMAL"
                            }
                        }
                    ]
                }
            },
            "size": size
        }

        result = self.search_companies(query)
        return [hit["_source"] for hit in result.get("hits", {}).get("hits", [])]

    def scroll_all_companies(self, query: Dict[str, Any], scroll_time: str = "2m") -> List[Dict[str, Any]]:
        """Use scroll API to get all results from a query"""
        all_results = []

        # Initial scroll request
        scroll_query = {**query, "size": 100}
        url = f"{self.config.base_url}/virksomhed/_search?scroll={scroll_time}"

        try:
            response = self.session.post(url, json=scroll_query, headers={'Content-Type': 'application/json'})
            response.raise_for_status()
            data = response.json()

            scroll_id = data.get("_scroll_id")
            hits = data.get("hits", {}).get("hits", [])
            all_results.extend([hit["_source"] for hit in hits])

            # Continue scrolling
            while hits and scroll_id:
                scroll_query = {
                    "scroll": scroll_time,
                    "scroll_id": scroll_id
                }

                response = self.session.post(
                    f"{self.config.base_url}/_search/scroll",
                    json=scroll_query,
                    headers={'Content-Type': 'application/json'}
                )
                response.raise_for_status()
                data = response.json()

                scroll_id = data.get("_scroll_id")
                hits = data.get("hits", {}).get("hits", [])
                all_results.extend([hit["_source"] for hit in hits])

                self.logger.info(f"Scrolled {len(all_results)} results so far...")

        except Exception as e:
            self.logger.error(f"Scroll error: {e}")
            raise

        return all_results

# Usage Example
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)

    client = CVRClient()

    try:
        # Search by CVR number
        company = client.get_company_by_cvr("10103940")
        print(f"Found company: {company['Vrvirksomhed']['navne'][0]['navn']}")

        # Search by name
        companies = client.search_companies_by_name("Novo Nordisk")
        print(f"Found {len(companies)} companies matching 'Novo Nordisk'")

        # Get companies in IT industry
        it_companies = client.get_active_companies_by_industry("620100")
        print(f"Found {len(it_companies)} active IT companies")

    except CVRError as e:
        print(f"CVR API error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")

# Comprehensive error handling
try:
    result = client.search_companies(query)
except CVRResultWindowError:
    # Use scroll API for large results
    result = client.scroll_all_companies(query)
except CVRAuthenticationError:
    # Refresh credentials or alert ops team
    logger.critical("Authentication failed - check credentials")
    raise
except CVRTimeoutError:
    # Retry with exponential backoff
    logger.warning("Request timed out - retrying")
    # Implement retry logic
except CVRError as e:
    # Log and handle gracefully
    logger.error(f"CVR API error: {e}")
    # Return cached data or default response

import time
from collections import deque
from threading import Lock

class RateLimiter:
    def __init__(self, max_requests=5, time_window=1.0):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = deque()
        self.lock = Lock()

    def acquire(self):
        with self.lock:
            now = time.time()

            # Remove old requests outside the time window
            while self.requests and self.requests[0] <= now - self.time_window:
                self.requests.popleft()

            # Check if we can make a request
            if len(self.requests) < self.max_requests:
                self.requests.append(now)
                return True

            # Wait until we can make a request
            sleep_time = self.requests[0] + self.time_window - now
            if sleep_time > 0:
                time.sleep(sleep_time)
                return self.acquire()

            return True

import redis
import json
from datetime import timedelta

class CVRCache:
    def __init__(self, redis_client=None):
        self.redis = redis_client or redis.Redis(
            host='localhost', 
            port=6379, 
            decode_responses=True
        )
        self.default_ttl = 3600  # 1 hour

    def get(self, key):
        try:
            data = self.redis.get(key)
            return json.loads(data) if data else None
        except (redis.RedisError, json.JSONDecodeError):
            return None

    def set(self, key, value, ttl=None):
        try:
            self.redis.setex(
                key, 
                ttl or self.default_ttl, 
                json.dumps(value, ensure_ascii=False)
            )
        except redis.RedisError:
            pass  # Continue without caching

    def get_or_fetch(self, key, fetch_func, ttl=None):
        # Try cache first
        cached = self.get(key)
        if cached:
            return cached

        # Fetch from API
        result = fetch_func()

        # Cache the result
        self.set(key, result, ttl)
        return result

import structlog
from prometheus_client import Counter, Histogram, Gauge
import time

# Metrics
REQUEST_COUNT = Counter('cvr_api_requests_total', 'Total CVR API requests', ['endpoint', 'status'])
REQUEST_DURATION = Histogram('cvr_api_request_duration_seconds', 'CVR API request duration')
ACTIVE_CONNECTIONS = Gauge('cvr_api_active_connections', 'Active CVR API connections')

logger = structlog.get_logger()

class MonitoredCVRClient(CVRClient):
    def _make_request(self, endpoint: str, query: Dict[str, Any]) -> Dict[str, Any]:
        start_time = time.time()
        ACTIVE_CONNECTIONS.inc()

        try:
            result = super()._make_request(endpoint, query)
            REQUEST_COUNT.labels(endpoint=endpoint, status='success').inc()
            return result

        except Exception as e:
            REQUEST_COUNT.labels(endpoint=endpoint, status='error').inc()
            logger.error(
                "CVR API request failed",
                endpoint=endpoint,
                query_size=len(json.dumps(query)),
                error=str(e)
            )
            raise

        finally:
            duration = time.time() - start_time
            REQUEST_DURATION.observe(duration)
            ACTIVE_CONNECTIONS.dec()

            logger.info(
                "CVR API request completed",
                endpoint=endpoint,
                duration=duration,
                query_size=len(json.dumps(query))
            )

import os
from dataclasses import dataclass, field
from typing import Dict, Any
import yaml

@dataclass
class CVRConfig:
    # API Configuration
    username: str = field(default_factory=lambda: os.getenv('CVR_USERNAME', ''))
    password: str = field(default_factory=lambda: os.getenv('CVR_PASSWORD', ''))
    base_url: str = field(default_factory=lambda: os.getenv('CVR_BASE_URL', 'http://distribution.virk.dk/cvr-permanent'))

    # Connection Settings
    timeout: int = field(default_factory=lambda: int(os.getenv('CVR_TIMEOUT', '30')))
    max_retries: int = field(default_factory=lambda: int(os.getenv('CVR_MAX_RETRIES', '3')))
    rate_limit_delay: float = field(default_factory=lambda: float(os.getenv('CVR_RATE_LIMIT_DELAY', '0.2')))

    # Cache Settings
    cache_ttl: int = field(default_factory=lambda: int(os.getenv('CVR_CACHE_TTL', '3600')))
    redis_url: str = field(default_factory=lambda: os.getenv('REDIS_URL', 'redis://localhost:6379'))

    @classmethod
    def from_file(cls, config_path: str):
        """Load configuration from YAML file"""
        with open(config_path, 'r') as f:
            config_data = yaml.safe_load(f)
        return cls(**config_data.get('cvr', {}))

    def validate(self):
        """Validate configuration"""
        if not self.username or not self.password:
            raise ValueError("CVR username and password are required")

        if self.timeout <= 0:
            raise ValueError("Timeout must be positive")

        if self.rate_limit_delay < 0:
            raise ValueError("Rate limit delay cannot be negative")

# Example config.yaml
"""
cvr:
  username: "${CVR_USERNAME}"
  password: "${CVR_PASSWORD}"
  base_url: "http://distribution.virk.dk/cvr-permanent"
  timeout: 30
  max_retries: 3
  rate_limit_delay: 0.2
  cache_ttl: 3600
  redis_url: "redis://localhost:6379"
"""

import unittest
from unittest.mock import Mock, patch
import pytest
from cvr_client import CVRClient, CVRError, CVRResultWindowError

class TestCVRClient(unittest.TestCase):
    def setUp(self):
        self.client = CVRClient()

    @patch('requests.Session.post')
    def test_get_company_by_cvr_success(self, mock_post):
        # Mock successful response
        mock_response = Mock()
        mock_response.status_code = 200
        mock_response.json.return_value = {
            "hits": {
                "hits": [
                    {
                        "_source": {
                            "Vrvirksomhed": {
                                "cvrNummer": 10103940,
                                "navne": [{"navn": "Statsministeriet"}]
                            }
                        }
                    }
                ]
            }
        }
        mock_post.return_value = mock_response

        result = self.client.get_company_by_cvr("10103940")

        self.assertIsNotNone(result)
        self.assertEqual(result["Vrvirksomhed"]["cvrNummer"], 10103940)

    @patch('requests.Session.post')
    def test_result_window_error(self, mock_post):
        # Mock result window error
        mock_response = Mock()
        mock_response.status_code = 500
        mock_response.json.return_value = {
            "error": {
                "reason": "Result window is too large"
            }
        }
        mock_post.return_value = mock_response

        with pytest.raises(CVRResultWindowError):
            self.client.search_companies({"from": 3001, "size": 1})

import pytest
from cvr_client import CVRClient
import time

class TestCVRIntegration:
    @pytest.fixture
    def client(self):
        return CVRClient()

    def test_real_api_connection(self, client):
        """Test actual API connection"""
        result = client.get_company_by_cvr("10103940")
        assert result is not None
        assert result["Vrvirksomhed"]["cvrNummer"] == 10103940

    def test_rate_limiting(self, client):
        """Test rate limiting works"""
        start_time = time.time()

        # Make multiple requests
        for _ in range(3):
            client.get_company_by_cvr("10103940")

        elapsed = time.time() - start_time
        # Should take at least 0.4 seconds (2 * 0.2s delay)
        assert elapsed >= 0.4

    @pytest.mark.parametrize("cvr_number,expected_name", [
        ("10103940", "Statsministeriet"),
        ("10150817", "NOVO NORDISK A/S"),
    ])
    def test_known_companies(self, client, cvr_number, expected_name):
        """Test known companies return expected results"""
        result = client.get_company_by_cvr(cvr_number)
        assert result is not None
        names = [n["navn"] for n in result["Vrvirksomhed"]["navne"]]
        assert any(expected_name.upper() in name.upper() for name in names)