Skip to content

table_bases

sc_crawler.table_bases #

Tiny helper classes for the most commonly used fields to be inherited by sc_crawler.tables.

Classes:

Name Description
ScMetaModel

Custom class factory to auto-update table models.
ScModel

Custom extensions to SQLModel objects and tables.
MetaColumns

Helper class to add the status and observed_at columns.
HasPriceFieldsBase
ServerFields
BenchmarkScoreFields

ScMetaModel #

Bases: __class__

Custom class factory to auto-update table models.

  • Reuse description of the table and its fields as SQL comment.

    Checking if the table and its fields have explicit comment set to be shown in the CREATE TABLE statements, and if not, reuse the optional table and field descriptions. Table docstrings are truncated to first line.

  • Reuse description of the fields to dynamically append to the docstring in the Attributes section.

  • Set __validator__ to the parent Pydantic model without table=True, which is useful for running validations. The Pydantic model is found by the parent class' name ending in "Base".

  • Auto-generate SCD table docs from the non-SCD table docs.

Source code in sc_crawler/table_bases.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class ScMetaModel(SQLModel.__class__):
    """Custom class factory to auto-update table models.

    - Reuse description of the table and its fields as SQL comment.

        Checking if the table and its fields have explicit comment set
        to be shown in the `CREATE TABLE` statements, and if not,
        reuse the optional table and field descriptions. Table
        docstrings are truncated to first line.

    - Reuse description of the fields to dynamically append to the
        docstring in the Attributes section.

    - Set `__validator__` to the parent Pydantic model without
        `table=True`, which is useful for running validations.
        The Pydantic model is found by the parent class' name ending in "Base".

    - Auto-generate SCD table docs from the non-SCD table docs.
    """

    def __init__(subclass, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # early return for non-tables
        if subclass.model_config.get("table") is None:
            return
        satable = subclass.metadata.tables[subclass.__tablename__]

        # enforce auto-naming constrains as per
        # https://alembic.sqlalchemy.org/en/latest/naming.html
        subclass.metadata.naming_convention = {
            "ix": "ix_%(column_0_label)s",
            "uq": "uq_%(table_name)s_%(column_0_name)s",
            "ck": "ck_%(table_name)s_%(constraint_name)s",
            "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
            "pk": "pk_%(table_name)s",
        }

        # table comment
        if subclass.__doc__ and satable.comment is None:
            satable.comment = subclass.__doc__.splitlines()[0]

        # column comments
        for k, v in subclass.model_fields.items():
            comment = satable.columns[k].comment
            if v.description and comment is None:
                satable.columns[k].comment = v.description

        # generate docstring for SCD tables
        if subclass.__name__.endswith("Scd"):
            from .tables import tables

            nonscd = [t for t in tables if t.__name__ == subclass.__name__[:-3]][0]
            doclines = nonscd.__doc__.splitlines()
            # drop trailing dot and append SCD
            doclines[0] = doclines[0][:-1] + " (SCD Type 2)."
            subclass.__doc__ = "\n".join(doclines)
        else:
            # describe table columns as attributes in docstring
            subclass.__doc__ = subclass.__doc__ + "\n\nAttributes:\n"
            for k, v in subclass.model_fields.items():
                if not hasattr(v.annotation, "__args__"):
                    typehint = v.annotation.__name__
                else:
                    typehint = str(v.annotation)
                description = satable.columns[k].comment
                subclass.__doc__ = (
                    subclass.__doc__ + f"    {k} ({typehint}): {description}\n"
                )

        # find Pydantic model parent to be used for validating
        subclass.__validator__ = [
            m for m in subclass.__bases__ if m.__name__.endswith("Base")
        ][0]

ScModel #

Bases: SQLModel

Custom extensions to SQLModel objects and tables.

Extra features:

  • auto-generated table names using snake_case,
  • support for hashing table rows,
  • reuse description field of tables/columns as SQL comment,
  • reuse description field of columns to extend the Attributes section of the docstring.

Methods:

Name Description
__tablename__

Override tables names using all-lowercase snake_case.
get_columns

Return the table's column names in a dict for all, primary keys, and attributes.
get_table_name

Return the SQLModel object's table name.
get_validator

Return the parent Base Pydantic model (without a table definition).
get_scd

Return the SCD version of the SQLModel table.
hash

Hash the content of the rows.
Source code in sc_crawler/table_bases.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class ScModel(SQLModel, metaclass=ScMetaModel):
    """Custom extensions to SQLModel objects and tables.

    Extra features:

    - auto-generated table names using [snake_case][sc_crawler.str_utils.snake_case],
    - support for hashing table rows,
    - reuse description field of tables/columns as SQL comment,
    - reuse description field of columns to extend the `Attributes` section of the docstring.
    """

    @declared_attr  # type: ignore
    def __tablename__(cls) -> str:
        """Override tables names using all-lowercase [snake_case][sc_crawler.str_utils.snake_case]."""
        return snake_case(cls.__name__)

    @classmethod
    def get_columns(cls) -> List[str]:
        """Return the table's column names in a dict for all, primary keys, and attributes."""
        columns = cls.__table__.columns.keys()
        pks = [pk.name for pk in inspect(cls).primary_key]
        attributes = [a for a in columns if a not in set(pks)]
        return {"all": columns, "primary_keys": pks, "attributes": attributes}

    @classmethod
    def get_table_name(cls) -> str:
        """Return the SQLModel object's table name."""
        return str(cls.__tablename__)

    @classmethod
    def get_validator(cls) -> Union["ScModel", None]:
        """Return the parent Base Pydantic model (without a table definition)."""
        if cls.model_config.get("table") is None:
            return None
        return cls.__validator__

    @classmethod
    def get_scd(cls) -> Union["ScModel", None]:
        """Return the SCD version of the SQLModel table."""
        if cls.model_config.get("table") is None:
            return None
        from .tables_scd import tables_scd

        validator = cls.get_validator()
        scds = [t for t in tables_scd if t.get_validator() == validator]
        if len(scds) != 1:
            raise ValueError("Not found SCD definition.")
        return scds[0]

    @classmethod
    def hash(
        cls,
        session: Session,
        ignored: List[str] = ["observed_at"],
        progress: Optional[Progress] = None,
    ) -> dict:
        """Hash the content of the rows.

        Args:
            session: Database connection to use for object lookups.
            ignored: List of column names to exclude from hashing.
            progress: Optional progress bar to track the status of the hashing.

        Returns:
            Dictionary of the row hashes keyed by the JSON dump of primary keys.
        """
        pks = sorted(cls.get_columns()["primary_keys"])
        rows = session.exec(statement=select(cls))
        row_count = session.query(cls).count()
        if progress:
            table_task_id = progress.add_task(
                cls.get_table_name(),
                total=row_count,
            )
        # no use of a generator as will need to serialize to JSON anyway
        hashes = {}
        for i, row in enumerate(rows):
            # NOTE Pydantic is warning when read Gpu/Storage as dict
            # https://github.com/tiangolo/sqlmodel/issues/63#issuecomment-1081555082
            rowdict = row.model_dump(warnings=False)
            keys = {pk: rowdict.get(pk) for pk in pks}
            keys_id = dumps(keys, sort_keys=True)
            for dropkey in [*ignored, *pks]:
                rowdict.pop(dropkey, None)
            rowhash = sha1(dumps(rowdict, sort_keys=True).encode()).hexdigest()
            hashes[keys_id] = rowhash
            if progress:
                # updating the progress bar is expensive, so limit with manu iterations
                if row_count > 1e3:
                    if (i + 1) % 1000 == 0:
                        progress.update(table_task_id, advance=1000)
                    if i == row_count - 1:
                        progress.update(table_task_id, advance=row_count % 1000)
                else:
                    progress.update(table_task_id, advance=1)

        return hashes

__tablename__ #

__tablename__()

Override tables names using all-lowercase snake_case.

Source code in sc_crawler/table_bases.py 
119
120
121
122
@declared_attr  # type: ignore
def __tablename__(cls) -> str:
    """Override tables names using all-lowercase [snake_case][sc_crawler.str_utils.snake_case]."""
    return snake_case(cls.__name__)

get_columns classmethod #

get_columns()

Return the table's column names in a dict for all, primary keys, and attributes.

Source code in sc_crawler/table_bases.py 
124
125
126
127
128
129
130
@classmethod
def get_columns(cls) -> List[str]:
    """Return the table's column names in a dict for all, primary keys, and attributes."""
    columns = cls.__table__.columns.keys()
    pks = [pk.name for pk in inspect(cls).primary_key]
    attributes = [a for a in columns if a not in set(pks)]
    return {"all": columns, "primary_keys": pks, "attributes": attributes}

get_table_name classmethod #

get_table_name()

Return the SQLModel object's table name.

Source code in sc_crawler/table_bases.py 
132
133
134
135
@classmethod
def get_table_name(cls) -> str:
    """Return the SQLModel object's table name."""
    return str(cls.__tablename__)

get_validator classmethod #

get_validator()

Return the parent Base Pydantic model (without a table definition).

Source code in sc_crawler/table_bases.py 
137
138
139
140
141
142
@classmethod
def get_validator(cls) -> Union["ScModel", None]:
    """Return the parent Base Pydantic model (without a table definition)."""
    if cls.model_config.get("table") is None:
        return None
    return cls.__validator__

get_scd classmethod #

get_scd()

Return the SCD version of the SQLModel table.

Source code in sc_crawler/table_bases.py 
144
145
146
147
148
149
150
151
152
153
154
155
@classmethod
def get_scd(cls) -> Union["ScModel", None]:
    """Return the SCD version of the SQLModel table."""
    if cls.model_config.get("table") is None:
        return None
    from .tables_scd import tables_scd

    validator = cls.get_validator()
    scds = [t for t in tables_scd if t.get_validator() == validator]
    if len(scds) != 1:
        raise ValueError("Not found SCD definition.")
    return scds[0]

hash classmethod #

hash(session, ignored=['observed_at'], progress=None)

Hash the content of the rows.

Parameters:

Name Type Description Default
session Session

Database connection to use for object lookups.

 required
ignored List[str]

List of column names to exclude from hashing.

 ['observed_at']
progress Optional[Progress]

Optional progress bar to track the status of the hashing.

 None

Returns:

Type Description
dict

Dictionary of the row hashes keyed by the JSON dump of primary keys.
Source code in sc_crawler/table_bases.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@classmethod
def hash(
    cls,
    session: Session,
    ignored: List[str] = ["observed_at"],
    progress: Optional[Progress] = None,
) -> dict:
    """Hash the content of the rows.

    Args:
        session: Database connection to use for object lookups.
        ignored: List of column names to exclude from hashing.
        progress: Optional progress bar to track the status of the hashing.

    Returns:
        Dictionary of the row hashes keyed by the JSON dump of primary keys.
    """
    pks = sorted(cls.get_columns()["primary_keys"])
    rows = session.exec(statement=select(cls))
    row_count = session.query(cls).count()
    if progress:
        table_task_id = progress.add_task(
            cls.get_table_name(),
            total=row_count,
        )
    # no use of a generator as will need to serialize to JSON anyway
    hashes = {}
    for i, row in enumerate(rows):
        # NOTE Pydantic is warning when read Gpu/Storage as dict
        # https://github.com/tiangolo/sqlmodel/issues/63#issuecomment-1081555082
        rowdict = row.model_dump(warnings=False)
        keys = {pk: rowdict.get(pk) for pk in pks}
        keys_id = dumps(keys, sort_keys=True)
        for dropkey in [*ignored, *pks]:
            rowdict.pop(dropkey, None)
        rowhash = sha1(dumps(rowdict, sort_keys=True).encode()).hexdigest()
        hashes[keys_id] = rowhash
        if progress:
            # updating the progress bar is expensive, so limit with manu iterations
            if row_count > 1e3:
                if (i + 1) % 1000 == 0:
                    progress.update(table_task_id, advance=1000)
                if i == row_count - 1:
                    progress.update(table_task_id, advance=row_count % 1000)
            else:
                progress.update(table_task_id, advance=1)

    return hashes

MetaColumns #

Bases: ScModel

Helper class to add the status and observed_at columns.

Source code in sc_crawler/table_bases.py 
207
208
209
210
211
212
213
214
215
216
217
218
class MetaColumns(ScModel):
    """Helper class to add the `status` and `observed_at` columns."""

    status: Status = Field(
        default=Status.ACTIVE,
        description="Status of the resource (active or inactive).",
    )
    observed_at: datetime = Field(
        default_factory=lambda: datetime.now(UTC),
        sa_column_kwargs={"onupdate": lambda: datetime.now(UTC)},
        description="Timestamp of the last observation.",
    )

HasPriceFieldsBase #

Bases: ScModel

Methods:

Name Description
serialize_price_tiers

Serialize price_tiered field, converting dicts to PriceTier instances first.
Source code in sc_crawler/table_bases.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
class HasPriceFieldsBase(ScModel):
    unit: PriceUnit = Field(description="Billing unit of the pricing model.")
    # set to max price if tiered
    price: float = Field(description="Actual price of a billing unit.")
    # e.g. setup fee for dedicated servers,
    # or upfront costs of a reserved instance type
    price_upfront: float = Field(
        default=0, description="Price to be paid when setting up the resource."
    )
    price_tiered: List[PriceTier] = Field(
        default=[],
        sa_type=JSON,
        description="List of pricing tiers with min/max thresholds and actual prices.",
    )
    currency: str = Field(default="USD", description="Currency of the prices.")

    @field_serializer("price_tiered")
    @classmethod
    def serialize_price_tiers(cls, value):
        """Serialize price_tiered field, converting dicts to PriceTier instances first."""
        if value is None:
            return []
        return [
            (PriceTier(**item) if isinstance(item, dict) else item) for item in value
        ]

serialize_price_tiers classmethod #

serialize_price_tiers(value)

Serialize price_tiered field, converting dicts to PriceTier instances first.

Source code in sc_crawler/table_bases.py 
342
343
344
345
346
347
348
349
350
@field_serializer("price_tiered")
@classmethod
def serialize_price_tiers(cls, value):
    """Serialize price_tiered field, converting dicts to PriceTier instances first."""
    if value is None:
        return []
    return [
        (PriceTier(**item) if isinstance(item, dict) else item) for item in value
    ]

ServerFields #

Bases: HasDescription, HasDisplayName, HasApiReference, HasName, HasServerIdPK, HasVendorPKFK

Methods:

Name Description
serialize_cpus

Serialize cpus field, converting dicts to Cpu instances first.
serialize_gpus

Serialize gpus field, converting dicts to Gpu instances first.
serialize_storages

Serialize storages field, converting dicts to Disk instances first.
Source code in sc_crawler/table_bases.py 
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
class ServerFields(
    HasDescription,
    HasDisplayName,
    HasApiReference,
    HasName,
    HasServerIdPK,
    HasVendorPKFK,
):
    family: Optional[str] = Field(
        default=None,
        description="Server family, e.g. General-purpose machine (GCP), or M5g (AWS).",
    )
    # inspector_status: str = Field(
    #     default=None,
    #     description="Server family, e.g. General-purpose machine (GCP), or M5g (AWS).",
    # )
    vcpus: int = Field(
        default=None,
        description="Default number of virtual CPUs (vCPU) of the server.",
    )
    hypervisor: Optional[str] = Field(
        default=None,
        description="Hypervisor of the virtual server, e.g. Xen, KVM, Nitro or Dedicated.",
    )
    cpu_allocation: CpuAllocation = Field(
        default=None,
        description="Allocation of CPU(s) to the server, e.g. shared, burstable or dedicated.",
    )
    cpu_cores: Optional[int] = Field(
        default=None,
        description=(
            "Default number of CPU cores of the server. "
            "Equals to vCPUs when HyperThreading is disabled."
        ),
    )
    cpu_speed: Optional[float] = Field(
        default=None, description="Vendor-reported maximum CPU clock speed (GHz)."
    )
    cpu_architecture: CpuArchitecture = Field(
        default=None,
        description="CPU architecture (arm64, arm64_mac, i386, or x86_64).",
    )
    cpu_manufacturer: Optional[str] = Field(
        default=None,
        description="The manufacturer of the primary processor, e.g. Intel or AMD.",
    )
    cpu_family: Optional[str] = Field(
        default=None,
        description="The product line/family of the primary processor, e.g. Xeon, Core i7, Ryzen 9.",
    )
    cpu_model: Optional[str] = Field(
        default=None,
        description="The model number of the primary processor, e.g. 9750H.",
    )
    cpu_l1_cache: Optional[int] = Field(
        default=None, description="L1 cache size (byte)."
    )
    cpu_l2_cache: Optional[int] = Field(
        default=None, description="L2 cache size (byte)."
    )
    cpu_l3_cache: Optional[int] = Field(
        default=None, description="L3 cache size (byte)."
    )
    cpu_flags: List[str] = Field(
        sa_type=JSON, default=[], description="CPU features/flags."
    )
    cpus: List[Cpu] = Field(
        default=[],
        sa_type=JSON,
        description=(
            "JSON array of known CPU details, e.g. the manufacturer, family, model; "
            "L1/L2/L3 cache size; microcode version; feature flags; bugs etc."
        ),
    )
    memory_amount: int = Field(
        default=None,
        description="RAM amount (MiB).",
    )
    memory_generation: Optional[DdrGeneration] = Field(
        default=None, description="Generation of the DDR SDRAM, e.g. DDR4 or DDR5."
    )
    memory_speed: Optional[int] = Field(
        default=None, description="DDR SDRAM clock rate (Mhz)."
    )
    memory_ecc: Optional[bool] = Field(
        default=None,
        description="If the DDR SDRAM uses error correction code to detect and correct n-bit data corruption.",
    )
    gpu_count: float = Field(
        default=0,
        description="Number of GPU accelerator(s).",
    )
    gpu_memory_min: Optional[int] = Field(
        default=None,
        description="Memory (MiB) allocated to the lowest-end GPU accelerator.",
    )
    gpu_memory_total: Optional[int] = Field(
        default=None,
        description="Overall memory (MiB) allocated to all the GPU accelerator(s).",
    )
    gpu_manufacturer: Optional[str] = Field(
        default=None,
        description="The manufacturer of the primary GPU accelerator, e.g. Nvidia or AMD.",
    )
    gpu_family: Optional[str] = Field(
        default=None,
        description="The product family of the primary GPU accelerator, e.g. Turing.",
    )
    gpu_model: Optional[str] = Field(
        default=None,
        description="The model number of the primary GPU accelerator, e.g. Tesla T4.",
    )
    gpus: List[Gpu] = Field(
        default=[],
        sa_type=JSON,
        description=(
            "JSON array of GPU accelerator details, including "
            "the manufacturer, name, and memory (MiB) of each GPU."
        ),
    )
    storage_size: int = Field(
        default=0,
        description="Overall size (GB) of the disk(s).",
    )
    storage_type: Optional[StorageType] = Field(
        default=None,
        description="Primary disk type, e.g. HDD, SSD, NVMe SSD, or network).",
    )
    storages: List[Disk] = Field(
        default=[],
        sa_type=JSON,
        description=(
            "JSON array of disks attached to the server, including "
            "the size (MiB) and type of each disk."
        ),
    )
    network_speed: Optional[float] = Field(
        default=None,
        description="The baseline network performance (Gbps) of the network card.",
    )
    inbound_traffic: float = Field(
        default=0,
        description="Amount of complimentary inbound traffic (GB) per month.",
    )
    outbound_traffic: float = Field(
        default=0,
        description="Amount of complimentary outbound traffic (GB) per month.",
    )
    ipv4: int = Field(
        default=0, description="Number of complimentary IPv4 address(es)."
    )

    @field_serializer("cpus")
    @classmethod
    def serialize_cpus(cls, value):
        """Serialize cpus field, converting dicts to Cpu instances first."""
        if value is None:
            return []
        return [(Cpu(**item) if isinstance(item, dict) else item) for item in value]

    @field_serializer("gpus")
    @classmethod
    def serialize_gpus(cls, value):
        """Serialize gpus field, converting dicts to Gpu instances first."""
        if value is None:
            return []
        return [(Gpu(**item) if isinstance(item, dict) else item) for item in value]

    @field_serializer("storages")
    @classmethod
    def serialize_storages(cls, value):
        """Serialize storages field, converting dicts to Disk instances first."""
        if value is None:
            return []
        return [(Disk(**item) if isinstance(item, dict) else item) for item in value]

serialize_cpus classmethod #

serialize_cpus(value)

Serialize cpus field, converting dicts to Cpu instances first.

Source code in sc_crawler/table_bases.py 
689
690
691
692
693
694
695
@field_serializer("cpus")
@classmethod
def serialize_cpus(cls, value):
    """Serialize cpus field, converting dicts to Cpu instances first."""
    if value is None:
        return []
    return [(Cpu(**item) if isinstance(item, dict) else item) for item in value]

serialize_gpus classmethod #

serialize_gpus(value)

Serialize gpus field, converting dicts to Gpu instances first.

Source code in sc_crawler/table_bases.py 
697
698
699
700
701
702
703
@field_serializer("gpus")
@classmethod
def serialize_gpus(cls, value):
    """Serialize gpus field, converting dicts to Gpu instances first."""
    if value is None:
        return []
    return [(Gpu(**item) if isinstance(item, dict) else item) for item in value]

serialize_storages classmethod #

serialize_storages(value)

Serialize storages field, converting dicts to Disk instances first.

Source code in sc_crawler/table_bases.py 
705
706
707
708
709
710
711
@field_serializer("storages")
@classmethod
def serialize_storages(cls, value):
    """Serialize storages field, converting dicts to Disk instances first."""
    if value is None:
        return []
    return [(Disk(**item) if isinstance(item, dict) else item) for item in value]

BenchmarkScoreFields #

Bases: HasBenchmarkPKFK, HasServerPK, HasVendorPKFK

Methods:

Name Description
update_config_to_hashable

We need a hashable column for the primary key.
Source code in sc_crawler/table_bases.py 
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
class BenchmarkScoreFields(HasBenchmarkPKFK, HasServerPK, HasVendorPKFK):
    model_config = ConfigDict(arbitrary_types_allowed=True)

    @model_validator(mode="before")
    def update_config_to_hashable(cls, values):
        """We need a hashable column for the primary key.

        Note that we also sort the keys, so that the resulting JSON
        can be compared as text as well (as some database engines do).
        """
        values["config"] = HashableDict(sorted(values.get("config", {}).items()))
        return values

    # use HashableDict as it's a primary key that needs to be hashable, but
    # fall back to dict to avoid PydanticInvalidForJsonSchema
    config: HashableDict | dict = Field(
        default={},
        sa_type=HashableJSON,
        primary_key=True,
        description='Dictionary of config parameters of the specific benchmark, e.g. {"bandwidth": 4096}',
    )
    score: float = Field(
        description="The resulting score of the benchmark.",
    )
    note: Optional[str] = Field(
        default=None,
        description="Optional note, comment or context on the benchmark score.",
    )

update_config_to_hashable #

update_config_to_hashable(values)

We need a hashable column for the primary key.

Note that we also sort the keys, so that the resulting JSON can be compared as text as well (as some database engines do).

Source code in sc_crawler/table_bases.py 
787
788
789
790
791
792
793
794
795
@model_validator(mode="before")
def update_config_to_hashable(cls, values):
    """We need a hashable column for the primary key.

    Note that we also sort the keys, so that the resulting JSON
    can be compared as text as well (as some database engines do).
    """
    values["config"] = HashableDict(sorted(values.get("config", {}).items()))
    return values