chore: restore files from backup

backend/scripts/test_herd_data.py

@@ -0,0 +1,46 @@
+import sys
+import os
+import logging
+from pathlib import Path
+
+# Add the project root to Python path
+project_root = str(Path(__file__).parents[2])
+sys.path.insert(0, project_root)
+
+from backend.src.sources.parsers.herd_data import HerdDataParser
+
+# Setup logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+def test_single_chr():
+    # Test configuration with direct secrets
+    config = {
+        'enabled': True,
+        'type': 'herd_data',
+        'project_id': 'landbrugsdata-1',  # Updated to correct project ID
+        'use_google_secrets': True,
+        'secrets': {
+            'fvm_username': '42731978',
+            'fvm_password': 'GiXOEqqWjzWa3BXWCPGT1SINA'
+        }
+    }
+
+    # Initialize parser
+    parser = HerdDataParser(config)
+
+    # Test with known CHR number
+    test_chr = "10860"
+
+    try:
+        result = parser.query_herd(test_chr)
+        logger.info(f"Result for CHR {test_chr}:")
+        logger.info(result)
+    except Exception as e:
+        logger.error(f"Error: {str(e)}")
+        logger.error(f"Error type: {type(e)}")
+        import traceback
+        logger.error(traceback.format_exc())
+
+if __name__ == "__main__":
+    test_single_chr() 

backend/src/base.py

@@ -4,22 +4,49 @@
 from shapely.geometry import shape
 import pyarrow as pa
 import logging
-from typing import Optional
+from typing import Optional, Dict, Any
 import time
 import os
 import pandas as pd
 from .sources.utils.geometry_validator import validate_and_transform_geometries
+from google.cloud import secretmanager
 
 logger = logging.getLogger(__name__)
 
 class Source(ABC):
     """Abstract base class for data sources"""
 
-    def __init__(self, config):
+    def __init__(self, config: Dict[str, Any]):
         self.config = config
+        self.enabled = config.get('enabled', True)
+        self.project_id = config.get('project_id')
+
+        self.use_google_secrets = config.get('use_google_secrets', False)
+        if self.use_google_secrets:
+            self.secrets_client = secretmanager.SecretManagerServiceClient()
+
         self.storage_client = storage.Client()
         self.bucket = self.storage_client.bucket('landbrugsdata-raw-data')
 
+    def get_secret(self, key: str) -> Optional[str]:
+        """Get secret from config, environment variables, or Google Secret Manager"""
+        if 'secrets' in self.config and key in self.config['secrets']:
+            return self.config['secrets'][key]
+
+        env_value = os.environ.get(key.upper())
+        if env_value:
+            return env_value
+
+        if self.use_google_secrets and self.project_id:
+            name = f"projects/{self.project_id}/secrets/{key}/versions/latest"
+            try:
+                response = self.secrets_client.access_secret_version(request={"name": name})
+                return response.payload.data.decode("UTF-8")
+            except Exception as e:
+                logger.error(f"Error accessing secret {key}: {str(e)}")
+
+        return None
+
     @abstractmethod
     async def fetch(self):
         """Fetch data from source"""