Method prepare

trinity/buffer/storage/sql.py:335–350

Initialize async engine and create tables.

(self)

            self.total_samples = float("inf")

    async def prepare(self) -> None:
        """Initialize async engine and create tables."""
        if self._initialized:
            return
        from trinity.buffer.schema.formatter import TaskFormatter

        result = await init_async_engine(
            self.config.path, self.config.name, self.config.schema_type  # type: ignore
        )
        self.engine, self.table_model_cls = result
        self.session = async_sessionmaker(self.engine, expire_on_commit=False)
        self.default_workflow_cls = WORKFLOWS.get(self.config.default_workflow_type)
        self.default_reward_fn_cls = REWARD_FUNCTIONS.get(self.config.default_reward_fn_type)
        self.formatter = TaskFormatter(self.config)
        self._initialized = True
        self.logger.info(f"SQL task storage initialized at {self.config.path}")

    async def write(self, data: List[Dict]) -> None:
        await self.prepare()