INNER CODE UNIT · Python

_ensure_db_pool

yihong1120/Construction-Hazard-Detection · main.py:70

    async def _ensure_db_pool(self) -> None:
        """Ensure a connection pool to the database is established before
        querying."""
        if self.db_pool is None:
            database_url = os.getenv('DATABASE_URL')
            if database_url is None:
                raise RuntimeError(
                    'DATABASE_URL environment variable is required',
                )
            url = make_url(database_url)
            if url.get_backend_name() != 'postgresql':
                raise RuntimeError(
                    'DATABASE_URL must use PostgreSQL',
                )
            port = url.port
            self.logger.info(
                '[database] Connecting to stream configuration database '
                '%s:%s/%s',

View source record →