INNER CODE UNIT · Python

fetch_stream_configs

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

    async def fetch_stream_configs(self) -> list[StreamConfig]:
        """Query the database for current stream configurations.

        Returns:
            list[StreamConfig]: All configured stream records.
        """
        await self._ensure_db_pool()
        if self.db_pool is None:
            raise RuntimeError(
                'Database pool is not initialized. Check DATABASE_URL and '
                'DB connectivity.',
            )
        sql = """
        SELECT sc.id               AS stream_id,
               sc.video_url,
               sc.updated_at,
               sc.model_key,
               s.name              AS site,

View source record →