def __init__()

in src/psearch/ingestion/services/spanner_service.py [0:0]

• 10 lines of code
• 3 McCabe index (conditional complexity)

    def __init__(self, project_id: str):
        """
        Initialize Spanner client and database connections

        Args:
            project_id: Google Cloud project ID
        """
        self.project_id = project_id
        self.instance_id = os.environ.get("SPANNER_INSTANCE_ID")
        self.database_id = os.environ.get("SPANNER_DATABASE_ID")

        if not self.instance_id or not self.database_id:
            raise ValueError("SPANNER_INSTANCE_ID and SPANNER_DATABASE_ID must be set")

        # Initialize Spanner client
        self.client = spanner.Client(project=project_id)
        self.instance = self.client.instance(self.instance_id)
        self.database = self.instance.database(self.database_id)

        logging.info(f"Initialized Spanner service for database: {self.database_id}")