| 67 | self.logger.info(f"SQL storage initialized at {self.config.path}") |
| 68 | |
| 69 | async def write(self, data: List[Experience]) -> None: |
| 70 | await self.prepare() |
| 71 | |
| 72 | async def operation(session: AsyncSession): |
| 73 | for exp in data: |
| 74 | exp_bytes = exp.serialize() |
| 75 | if ( |
| 76 | self.max_experience_bytes > 0 |
| 77 | and exp_bytes is not None |
| 78 | and len(exp_bytes) > self.max_experience_bytes |
| 79 | ): |
| 80 | self.logger.warning( |
| 81 | f"Experience size {len(exp_bytes)} bytes exceeds " |
| 82 | f"max_experience_bytes {self.max_experience_bytes}, skipping." |
| 83 | ) |
| 84 | continue |
| 85 | meta_row = self.table_model_cls.from_experience(exp) |
| 86 | session.add(meta_row) |
| 87 | await session.flush() |
| 88 | blob_row = self.blob_model_cls(id=meta_row.id, experience_bytes=exp_bytes) |
| 89 | session.add(blob_row) |
| 90 | |
| 91 | await async_run_with_retry_session( |
| 92 | self.session, operation, self.max_retry_times, self.max_retry_interval |
| 93 | ) |
| 94 | self.logger.info(f"Write {len(data)} experiences to SQL storage.") |
| 95 | |
| 96 | async def _fetch_blobs(self, session: AsyncSession, ids: List[int]) -> Dict[int, bytes]: |
| 97 | stmt = select(self.blob_model_cls).where(self.blob_model_cls.id.in_(ids)) |