Function sync_model_schema_data

backend/app/services/model/model_schema.py:35–112

()

async def sync_model_schema_data():
    global _provider_checksum, _model_schema_checksum, _i18n_checksum
    async with aiohttp.ClientSession() as session:
        response = await session.get(
            f"{CONFIG.TASKINGAI_INFERENCE_URL}/v1/cache_checksums",
        )
        response_wrapper = ResponseWrapper(response.status, await response.json())
        check_http_error(response_wrapper)
        response_data = response_wrapper.json()["data"]
        provider_checksum = response_data["provider_checksum"]
        model_schema_checksum = response_data["model_schema_checksum"]
        i18n_checksum = response_data["i18n_checksum"]
        if (
            provider_checksum == _provider_checksum
            and model_schema_checksum == _model_schema_checksum
            and i18n_checksum == _i18n_checksum
        ):
            logger.debug(f"Checksums are the same, no need to sync model schema data.")
            return

    async with aiohttp.ClientSession() as session:
        response = await session.get(
            f"{CONFIG.TASKINGAI_INFERENCE_URL}/v1/caches",
        )
        response_wrapper = ResponseWrapper(response.status, await response.json())
        check_http_error(response_wrapper)
        response_data = response_wrapper.json()["data"]

        model_schemas = [ModelSchema.build(model_schema_data) for model_schema_data in response_data["model_schemas"]]
        model_schemas.sort(key=lambda x: x.model_schema_id)

        providers = []
        for provider_data in response_data["providers"]:
            num_model_schemas = len(
                [
                    model_schema
                    for model_schema in model_schemas
                    if model_schema.provider_id == provider_data["provider_id"]
                ]
            )
            model_types = list(
                set(
                    [
                        model_schema.type
                        for model_schema in model_schemas
                        if model_schema.provider_id == provider_data["provider_id"]
                        # and model_schema.type != ModelType.WILDCARD
                    ]
                )
            )
            provider = Provider.build(provider_data, num_model_schemas, model_types)
            providers.append(provider)

        # put pinned providers at the front and sort the rest by provider_id
        providers.sort(key=sort_providers)

        i18n_dict = response_data["i18n"]