Method store_credential

src/storage/psql_manager.py:339–380

存储或更新凭证

(self, filename: str, credential_data: Dict[str, Any], mode: str = "geminicli")

    # ============ StorageBackend 协议方法 ============

    async def store_credential(self, filename: str, credential_data: Dict[str, Any], mode: str = "geminicli") -> bool:
        """存储或更新凭证"""
        self._ensure_initialized()
        filename = os.path.basename(filename)

        try:
            table_name = self._get_table_name(mode)
            async with self._pool.acquire() as conn:
                existing = await conn.fetchrow(
                    f"SELECT rotation_order FROM {table_name} WHERE filename = $1", filename
                )

                if existing:
                    await conn.execute(
                        f"""
                        UPDATE {table_name}
                        SET credential_data = $1,
                            updated_at = EXTRACT(EPOCH FROM NOW())
                        WHERE filename = $2
                        """,
                        json.dumps(credential_data), filename
                    )
                else:
                    row = await conn.fetchrow(
                        f"SELECT COALESCE(MAX(rotation_order), -1) + 1 AS next_order FROM {table_name}"
                    )
                    next_order = row["next_order"]
                    await conn.execute(
                        f"""
                        INSERT INTO {table_name}
                        (filename, credential_data, rotation_order, last_success)
                        VALUES ($1, $2, $3, $4)
                        """,
                        filename, json.dumps(credential_data), next_order, time.time()
                    )

            log.debug(f"Stored credential: {filename} (mode={mode})")
            return True

        except Exception as e:
            log.error(f"Error storing credential {filename}: {e}")
            return False

    async def get_credential(self, filename: str, mode: str = "geminicli") -> Optional[Dict[str, Any]]:
        """获取凭证数据"""