Skip to main content

schema

Classes concerning data schemas.

Classes​

BitfountSchema​

class BitfountSchema(    name: str,    description: str | None = None,    column_descriptions: Mapping[str, str] | None = None,):

A schema that defines the tables of a BaseSource.

It includes the table found in BaseSource and its features.

Arguments

  • **kwargs: Optional keyword arguments to be provided to _add_dataframe_features.
  • column_descriptions: A dictionary of column names and their descriptions.
  • description: The description of the datasource.
  • name: The name of the datasource associated with this schema.

Variables​

  • hash : str - The hash of this schema.

    This relates to the BaseSource(s) that were used in the generation of this schema to assure that this schema is used against compatible data sources.

    Returns: A sha256 hash of the _datasource_hashes.

Methods​

add_dataframe_features​

def add_dataframe_features(    self,    data: pd.DataFrame,    ignore_cols: Sequence[str] | None = None,    force_stypes: MutableMapping[_ForceStypeValue | _SemanticTypeValue, list[str]] | None = None,    column_descriptions: Mapping[str, str] | None = None,    *,    from_schema_generation: bool = False,) ‑> None:

Add the features of a dataframe to the schema.

This method is not called directly, but used as a hook in for yield_data in the BaseSource class.

When from_schema_generation is True (e.g. called from SchemaGenerationFromYieldData), the schema is updated from the data including number_of_records. When False, updates are skipped if schema_type is already "full" so the schema is not mutated during training.

add_feature​

def add_feature(self, feature_name: str, semantic_type: SemanticType, dtype: Any) ‑> None:

Add a single feature to the schema.

Note that this method does not support Categorical features.

Arguments

  • feature_name: The name of the feature.
  • semantic_type: The semantic type of the feature.
  • dtype: The dtype of the feature.

apply​

def apply(    self,    dataframe: pd.DataFrame,    keep_cols: list[str] | None = None,    image_cols: list[str] | None = None,) ‑> pandas.core.frame.DataFrame:

Applies the schema to a dataframe and returns the transformed dataframe.

Sequentially adds missing columns to the dataframe, removes superfluous columns from the dataframe, changes the types of the columns in the dataframe and finally encodes the categorical columns in the dataframe before returning the transformed dataframe.

Arguments

  • dataframe: The dataframe to transform.
  • keep_cols: A list of columns to keep even if they are not part of the schema. Defaults to None.
  • image_cols: The list of image columns in the dataframe. Defaults to None.

Returns The dataframe with the transformations applied.

decode_categorical​

def decode_categorical(self, feature: str, value: int) ‑> Any:

Decode label corresponding to a categorical feature in the schema.

Arguments

  • feature: The name of the feature.
  • value: The encoded value.

Returns The decoded feature value.

Raises

  • ValueError: If the feature cannot be found in the schema.
  • ValueError: If the label cannot be found in the feature encoder.

dump​

def dump(self, file_path: PathLike[Any]) ‑> None:

Dumps the schema as a yaml file.

Arguments

  • file_path: The path where the file should be saved

Returns none

dumps​

def dumps(self) ‑> str:

Produces the YAML representation of the schema object.

Returns The YAML representation of the schema as a string.

generate_full_schema​

def generate_full_schema(    self,    datasource: BaseSource,    force_stypes: MutableMapping[_ForceStypeValue | _SemanticTypeValue, list[str]] | None = None,    ignore_cols: list[str] | None = None,    secrets: APIKeys | RefreshableJWT | dict[SecretsUse, APIKeys | RefreshableJWT] | None = None,    *,    username: str | None = None,    reset_number_of_records: bool = True,) ‑> None:

Generate a full schema from a datasource.

Arguments

  • datasource: Data source to iterate for schema generation.
  • force_stypes: Optional forced semantic-type mapping.
  • ignore_cols: Optional columns to ignore.
  • secrets: Optional secrets for telemetry reporting.
  • username: Optional username for telemetry reporting. Required for JWT-based metrics reporting to authenticate successfully.
  • reset_number_of_records: Whether to reset number_of_records before processing. Set to False for incremental augmentation after a prior schema-generation phase (e.g. cache-first + uncached diff).

generate_optimized_full_schema​

def generate_optimized_full_schema(    self,    datasource: BaseSource,    force_stypes: MutableMapping[_ForceStypeValue | _SemanticTypeValue, list[str]] | None = None,    ignore_cols: list[str] | None = None,    secrets: APIKeys | RefreshableJWT | dict[SecretsUse, APIKeys | RefreshableJWT] | None = None,) ‑> None:

Generate full schema via cache, then process uncached files only.

Falls back to full yield-based generation when cache is not available for the datasource/cache state.

generate_partial_schema​

def generate_partial_schema(self, datasource: BaseSource) ‑> None:

Adds one batch of data to the schema.

generate_schema_from_cache​

def generate_schema_from_cache(    self,    datasource: BaseSource,    force_stypes: MutableMapping[_ForceStypeValue | _SemanticTypeValue, list[str]] | None = None,    ignore_cols: list[str] | None = None,    file_paths: Sequence[str | PathLike[str]] | None = None,    *,    sample_limit: int = 100,    reset_number_of_records: bool = False,) ‑> None:

Generate schema features directly from datasource cache.

This avoids loading full cached datasets into DataFrames when only schema metadata is needed.

Arguments

  • datasource: Data source whose cache should be queried.
  • force_stypes: Optional forced semantic-type mapping.
  • ignore_cols: Optional columns to ignore.
  • file_paths: Optional subset of file paths to scope cache queries to.
  • sample_limit: Maximum rows to sample for dtype reconciliation.
  • reset_number_of_records: Whether to reset number_of_records before applying cached row counts.

get_categorical_feature_size​

def get_categorical_feature_size(self, var: str | list[str]) ‑> int:

Gets the column dimensions.

Arguments

  • var: A column name or a list of column names for which to get the dimensions.

Returns The number of unique value in the categorical column.

get_categorical_feature_sizes​

def get_categorical_feature_sizes(    self, ignore_cols: str | list[str] | None = None,) ‑> list[int]:

Returns a list of categorical feature sizes.

Arguments

  • ignore_cols: The column(s) to be ignored from the schema.

get_column_names​

def get_column_names(    self, dataframe: pd.DataFrame, ignore_cols: list[str],) ‑> collections.abc.Iterable[str]:

Get the column names of the datasource.

get_feature_names​

def get_feature_names(self, semantic_type: SemanticType | None = None) ‑> list[str]:

Returns the names of all the features in the schema.

Arguments

  • semantic_type: if semantic type is provided, only the feature names corresponding to the semantic type are returned. Defaults to None.

Returns features: A list of feature names.

get_num_categorical​

def get_num_categorical(self, ignore_cols: str | list[str] | None = None) ‑> int:

Get the number of (non-ignored) categorical features.

Arguments

  • ignore_cols: Columns to ignore when counting categorical features.

get_num_continuous​

def get_num_continuous(self, ignore_cols: str | list[str] | None = None) ‑> int:

Get the number of (non-ignored) continuous features.

Arguments

  • ignore_cols: Columns to ignore when counting continuous features.

initialize_dataless_schema​

def initialize_dataless_schema(self, required_fields: dict[str, Any]) ‑> None:

Initialize the schema with required fields but no data.

Arguments

  • required_fields: A dictionary with field names and their types.

to_json​

def to_json(self) ‑> dict[str, typing.Any]:

Turns a schema object into a JSON compatible dictionary.

Returns A simple JSON compatible representation of the Schema

SchemaGenerationFromYieldData​

class SchemaGenerationFromYieldData(    schema: BitfountSchema,    ignore_cols: list[str] | None = None,    force_stypes: "MutableMapping[Literal['categorical', 'continuous', 'image', 'text', 'image_prefix'], list[str]] | None" = None,    secrets: APIKeys | RefreshableJWT | None = None,    is_file_iterable: bool = False,    username: str | None = None,):

Custom hook to execute logic during datasource yield data.

Initialize the hook.

Arguments

  • schema: The schema to update.
  • ignore_cols: Columns to ignore when updating the schema.
  • force_stypes: Forced semantic types for specific columns.
  • secrets: Secrets for authenticating with Bitfount services.
  • is_file_iterable: Whether the datasource is a file-iterable source. If False, diagnostic logs will only be emitted once per task.
  • username: Username for authenticating with Bitfount services. Required for JWT-based metrics reporting; without it session creation fails with an AuthenticatedUserError on every batch.

Ancestors​

Methods​

on_datasource_yield_data​

def on_datasource_yield_data(self, data: pd.DataFrame, *args: Any, **kwargs: Any) ‑> None:

Hook method triggered when the datasource yields data.

Arguments

  • data: The dataframe yielded by the datasource.
  • args: Additional arguments.
  • kwargs: Additional keyword arguments.

SchemaLevel​

class SchemaLevel(level: str | None):

Ordered representation of schema completeness levels.

Supports comparison operators: empty < partial < full.

Can be constructed from a string or None (treated as "empty"): SchemaLevel("full") > SchemaLevel("partial") # True SchemaLevel("empty") < SchemaLevel("full") # True SchemaLevel("partial") == SchemaLevel("partial") # True