lomas_core.models package

Submodules

lomas_core.models.collections module

class lomas_core.models.collections.BooleanMetadata(*, private_id: bool = False, nullable: bool = False, max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, type: Literal[MetadataColumnType.BOOLEAN])

Bases: ColumnMetadata

Model for boolean column metadata.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'max_influenced_partitions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_contributions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_length': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'nullable': FieldInfo(annotation=bool, required=False, default=False), 'private_id': FieldInfo(annotation=bool, required=False, default=False), 'type': FieldInfo(annotation=Literal[<MetadataColumnType.BOOLEAN: 'boolean'>], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: Literal[MetadataColumnType.BOOLEAN]

class lomas_core.models.collections.BoundedColumnMetadata(*, private_id: bool = False, nullable: bool = False, max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None)

Bases: ColumnMetadata

Model for columns with bounded data.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'max_influenced_partitions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_contributions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_length': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'nullable': FieldInfo(annotation=bool, required=False, default=False), 'private_id': FieldInfo(annotation=bool, required=False, default=False)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

validate_bounds()

Validates column bounds.

class lomas_core.models.collections.CategoricalColumnMetadata(*, private_id: bool = False, nullable: bool = False, max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None)

Bases: ColumnMetadata

Model for categorical column metadata.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'max_influenced_partitions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_contributions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_length': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'nullable': FieldInfo(annotation=bool, required=False, default=False), 'private_id': FieldInfo(annotation=bool, required=False, default=False)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

validate_categories()

Makes sure number of categories matches cardinality.

class lomas_core.models.collections.ColumnMetadata(*, private_id: bool = False, nullable: bool = False, max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None)

Bases: BaseModel

Base model for column metadata.

max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None

max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None

max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'max_influenced_partitions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_contributions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_length': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'nullable': FieldInfo(annotation=bool, required=False, default=False), 'private_id': FieldInfo(annotation=bool, required=False, default=False)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

nullable: bool

private_id: bool

class lomas_core.models.collections.DSAccess(*, database_type: str)

Bases: BaseModel

BaseModel for access info to a private dataset.

database_type: str

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'database_type': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.collections.DSInfo(*, dataset_name: str, dataset_access: DSPathAccess | DSS3Access, metadata_access: DSPathAccess | DSS3Access)

Bases: BaseModel

BaseModel for a dataset.

dataset_access: Annotated[DSPathAccess | DSS3Access, FieldInfo(annotation=NoneType, required=True, discriminator='database_type')]

dataset_name: str

metadata_access: Annotated[DSPathAccess | DSS3Access, FieldInfo(annotation=NoneType, required=True, discriminator='database_type')]

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_access': FieldInfo(annotation=Union[DSPathAccess, DSS3Access], required=True, discriminator='database_type'), 'dataset_name': FieldInfo(annotation=str, required=True), 'metadata_access': FieldInfo(annotation=Union[DSPathAccess, DSS3Access], required=True, discriminator='database_type')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.collections.DSPathAccess(*, database_type: Literal[PrivateDatabaseType.PATH], path: str)

Bases: DSAccess

BaseModel for a local dataset.

database_type: Literal[PrivateDatabaseType.PATH]

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'database_type': FieldInfo(annotation=Literal[<PrivateDatabaseType.PATH: 'PATH_DB'>], required=True), 'path': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

path: str

class lomas_core.models.collections.DSS3Access(*, database_type: Literal[PrivateDatabaseType.S3], endpoint_url: str, bucket: str, key: str, access_key_id: str | None = None, secret_access_key: str | None = None, credentials_name: str)

Bases: DSAccess

BaseModel for a dataset on S3.

access_key_id: str | None

bucket: str

credentials_name: str

database_type: Literal[PrivateDatabaseType.S3]

endpoint_url: str

key: str

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'access_key_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'bucket': FieldInfo(annotation=str, required=True), 'credentials_name': FieldInfo(annotation=str, required=True), 'database_type': FieldInfo(annotation=Literal[<PrivateDatabaseType.S3: 'S3_DB'>], required=True), 'endpoint_url': FieldInfo(annotation=str, required=True), 'key': FieldInfo(annotation=str, required=True), 'secret_access_key': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

secret_access_key: str | None

class lomas_core.models.collections.DatasetOfUser(*, dataset_name: str, initial_epsilon: float, initial_delta: float, total_spent_epsilon: float, total_spent_delta: float)

Bases: BaseModel

BaseModel for informations of a user on a dataset.

dataset_name: str

initial_delta: float

initial_epsilon: float

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True), 'initial_delta': FieldInfo(annotation=float, required=True), 'initial_epsilon': FieldInfo(annotation=float, required=True), 'total_spent_delta': FieldInfo(annotation=float, required=True), 'total_spent_epsilon': FieldInfo(annotation=float, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

total_spent_delta: float

total_spent_epsilon: float

class lomas_core.models.collections.DatasetsCollection(*, datasets: List[DSInfo])

Bases: BaseModel

BaseModel for datasets collection.

datasets: List[DSInfo]

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'datasets': FieldInfo(annotation=List[DSInfo], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.collections.DatetimeMetadata(*, private_id: bool = False, nullable: bool = False, max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, type: Literal[MetadataColumnType.DATETIME], lower: datetime, upper: datetime)

Bases: BoundedColumnMetadata

Model for datetime column metadata.

lower: datetime

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'lower': FieldInfo(annotation=datetime, required=True), 'max_influenced_partitions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_contributions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_length': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'nullable': FieldInfo(annotation=bool, required=False, default=False), 'private_id': FieldInfo(annotation=bool, required=False, default=False), 'type': FieldInfo(annotation=Literal[<MetadataColumnType.DATETIME: 'datetime'>], required=True), 'upper': FieldInfo(annotation=datetime, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: Literal[MetadataColumnType.DATETIME]

upper: datetime

class lomas_core.models.collections.FloatMetadata(*, private_id: bool = False, nullable: bool = False, max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, type: Literal[MetadataColumnType.FLOAT], precision: Precision, lower: float, upper: float)

Bases: BoundedColumnMetadata

Model for float column metadata.

lower: float

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'lower': FieldInfo(annotation=float, required=True), 'max_influenced_partitions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_contributions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_length': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'nullable': FieldInfo(annotation=bool, required=False, default=False), 'precision': FieldInfo(annotation=Precision, required=True), 'private_id': FieldInfo(annotation=bool, required=False, default=False), 'type': FieldInfo(annotation=Literal[<MetadataColumnType.FLOAT: 'float'>], required=True), 'upper': FieldInfo(annotation=float, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

precision: Precision

type: Literal[MetadataColumnType.FLOAT]

upper: float

class lomas_core.models.collections.IntCategoricalMetadata(*, private_id: bool = False, nullable: bool = False, max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, type: Literal[MetadataColumnType.INT], precision: Precision, cardinality: int, categories: List[int])

Bases: CategoricalColumnMetadata

Model for integer categorical column metadata.

cardinality: int

categories: List[int]

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'cardinality': FieldInfo(annotation=int, required=True), 'categories': FieldInfo(annotation=List[int], required=True), 'max_influenced_partitions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_contributions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_length': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'nullable': FieldInfo(annotation=bool, required=False, default=False), 'precision': FieldInfo(annotation=Precision, required=True), 'private_id': FieldInfo(annotation=bool, required=False, default=False), 'type': FieldInfo(annotation=Literal[<MetadataColumnType.INT: 'int'>], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

precision: Precision

type: Literal[MetadataColumnType.INT]

class lomas_core.models.collections.IntMetadata(*, private_id: bool = False, nullable: bool = False, max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, type: Literal[MetadataColumnType.INT], precision: Precision, lower: int, upper: int)

Bases: BoundedColumnMetadata

Model for integer column metadata.

lower: int

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'lower': FieldInfo(annotation=int, required=True), 'max_influenced_partitions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_contributions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_length': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'nullable': FieldInfo(annotation=bool, required=False, default=False), 'precision': FieldInfo(annotation=Precision, required=True), 'private_id': FieldInfo(annotation=bool, required=False, default=False), 'type': FieldInfo(annotation=Literal[<MetadataColumnType.INT: 'int'>], required=True), 'upper': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

precision: Precision

type: Literal[MetadataColumnType.INT]

upper: int

class lomas_core.models.collections.Metadata(*, max_ids: Annotated[int, Gt(gt=0)], rows: Annotated[int, Gt(gt=0)], row_privacy: bool, censor_dims: bool | None = False, columns: Dict[str, Annotated[Annotated[StrMetadata, Tag(tag=string)] | Annotated[StrCategoricalMetadata, Tag(tag=categorical_string)] | Annotated[IntMetadata, Tag(tag=int)] | Annotated[IntCategoricalMetadata, Tag(tag=categorical_int)] | Annotated[FloatMetadata, Tag(tag=float)] | Annotated[BooleanMetadata, Tag(tag=boolean)] | Annotated[DatetimeMetadata, Tag(tag=datetime)], Discriminator(discriminator=get_column_metadata_discriminator, custom_error_type=None, custom_error_message=None, custom_error_context=None)]])

Bases: BaseModel

BaseModel for a metadata format.

censor_dims: bool | None

columns: Dict[str, Annotated[Annotated[StrMetadata, Tag(tag=string)] | Annotated[StrCategoricalMetadata, Tag(tag=categorical_string)] | Annotated[IntMetadata, Tag(tag=int)] | Annotated[IntCategoricalMetadata, Tag(tag=categorical_int)] | Annotated[FloatMetadata, Tag(tag=float)] | Annotated[BooleanMetadata, Tag(tag=boolean)] | Annotated[DatetimeMetadata, Tag(tag=datetime)], Discriminator(discriminator=get_column_metadata_discriminator, custom_error_type=None, custom_error_message=None, custom_error_context=None)]]

max_ids: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])]

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'censor_dims': FieldInfo(annotation=Union[bool, NoneType], required=False, default=False), 'columns': FieldInfo(annotation=Dict[str, Annotated[Union[Annotated[StrMetadata, Tag], Annotated[StrCategoricalMetadata, Tag], Annotated[IntMetadata, Tag], Annotated[IntCategoricalMetadata, Tag], Annotated[FloatMetadata, Tag], Annotated[BooleanMetadata, Tag], Annotated[DatetimeMetadata, Tag]], Discriminator]], required=True), 'max_ids': FieldInfo(annotation=int, required=True, metadata=[Gt(gt=0)]), 'row_privacy': FieldInfo(annotation=bool, required=True), 'rows': FieldInfo(annotation=int, required=True, metadata=[Gt(gt=0)])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

row_privacy: bool

rows: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])]

class lomas_core.models.collections.StrCategoricalMetadata(*, private_id: bool = False, nullable: bool = False, max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, type: Literal[MetadataColumnType.STRING], cardinality: int, categories: List[str])

Bases: CategoricalColumnMetadata

Model for categorical string metadata.

cardinality: int

categories: List[str]

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'cardinality': FieldInfo(annotation=int, required=True), 'categories': FieldInfo(annotation=List[str], required=True), 'max_influenced_partitions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_contributions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_length': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'nullable': FieldInfo(annotation=bool, required=False, default=False), 'private_id': FieldInfo(annotation=bool, required=False, default=False), 'type': FieldInfo(annotation=Literal[<MetadataColumnType.STRING: 'string'>], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: Literal[MetadataColumnType.STRING]

class lomas_core.models.collections.StrMetadata(*, private_id: bool = False, nullable: bool = False, max_partition_length: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_influenced_partitions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, max_partition_contributions: Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])] | None = None, type: Literal[MetadataColumnType.STRING])

Bases: ColumnMetadata

Model for string metadata.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'max_influenced_partitions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_contributions': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'max_partition_length': FieldInfo(annotation=Union[Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Gt(gt=0)])], NoneType], required=False, default=None), 'nullable': FieldInfo(annotation=bool, required=False, default=False), 'private_id': FieldInfo(annotation=bool, required=False, default=False), 'type': FieldInfo(annotation=Literal[<MetadataColumnType.STRING: 'string'>], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: Literal[MetadataColumnType.STRING]

class lomas_core.models.collections.User(*, user_name: str, may_query: bool, datasets_list: List[DatasetOfUser])

Bases: BaseModel

BaseModel for a user in a user collection.

datasets_list: List[DatasetOfUser]

may_query: bool

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'datasets_list': FieldInfo(annotation=List[DatasetOfUser], required=True), 'may_query': FieldInfo(annotation=bool, required=True), 'user_name': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

user_name: str

class lomas_core.models.collections.UserCollection(*, users: List[User])

Bases: BaseModel

BaseModel for users collection.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'users': FieldInfo(annotation=List[User], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

users: List[User]

lomas_core.models.collections.get_column_metadata_discriminator(v: Any) → str

Discriminator function for determining the type of column metadata.

Parameters:

v (Any) – The unparsed column metadata (either dict or class object)

Raises:

ValueError – If the column type cannot be found.

Returns:

The metadata string type.

Return type:

str

lomas_core.models.config module

class lomas_core.models.config.Config(*, develop_mode: bool, server: Server, submit_limit: float, admin_database: MongoDBConfig | YamlDBConfig, private_db_credentials: List[S3CredentialsConfig], dp_libraries: DPLibraryConfig)

Bases: BaseModel

Server runtime config.

admin_database: Annotated[MongoDBConfig | YamlDBConfig, FieldInfo(annotation=NoneType, required=True, discriminator='db_type')]

develop_mode: bool

dp_libraries: DPLibraryConfig

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'admin_database': FieldInfo(annotation=Union[MongoDBConfig, YamlDBConfig], required=True, discriminator='db_type'), 'develop_mode': FieldInfo(annotation=bool, required=True), 'dp_libraries': FieldInfo(annotation=DPLibraryConfig, required=True), 'private_db_credentials': FieldInfo(annotation=List[S3CredentialsConfig], required=True, discriminator='db_type'), 'server': FieldInfo(annotation=Server, required=True), 'submit_limit': FieldInfo(annotation=float, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

private_db_credentials: List[S3CredentialsConfig]

server: Server

submit_limit: float

class lomas_core.models.config.DBConfig

Bases: BaseModel

BaseModel for database type config.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.config.DPLibraryConfig(*, opendp: OpenDPConfig)

Bases: BaseModel

BaseModel for DP librairies config.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'opendp': FieldInfo(annotation=OpenDPConfig, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

opendp: OpenDPConfig

class lomas_core.models.config.MongoDBConfig(*, db_type: Literal[AdminDBType.MONGODB], address: str, port: int, username: str, password: str, db_name: str, max_pool_size: int, min_pool_size: int, max_connecting: int)

Bases: DBConfig

BaseModel for dataset store configs in case of a MongoDB database.

address: str

db_name: str

db_type: Literal[AdminDBType.MONGODB]

max_connecting: int

max_pool_size: int

min_pool_size: int

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'address': FieldInfo(annotation=str, required=True), 'db_name': FieldInfo(annotation=str, required=True), 'db_type': FieldInfo(annotation=Literal[<AdminDBType.MONGODB: 'mongodb'>], required=True), 'max_connecting': FieldInfo(annotation=int, required=True), 'max_pool_size': FieldInfo(annotation=int, required=True), 'min_pool_size': FieldInfo(annotation=int, required=True), 'password': FieldInfo(annotation=str, required=True), 'port': FieldInfo(annotation=int, required=True), 'username': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

password: str

port: int

username: str

class lomas_core.models.config.OpenDPConfig(*, contrib: bool, floating_point: bool, honest_but_curious: bool)

Bases: BaseModel

BaseModel for openDP librairy config.

contrib: bool

floating_point: bool

honest_but_curious: bool

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'contrib': FieldInfo(annotation=bool, required=True), 'floating_point': FieldInfo(annotation=bool, required=True), 'honest_but_curious': FieldInfo(annotation=bool, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.config.PrivateDBCredentials

Bases: BaseModel

BaseModel for private database credentials.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.config.S3CredentialsConfig(*, db_type: Literal[PrivateDatabaseType.S3], credentials_name: str, access_key_id: str, secret_access_key: str, **extra_data: Any)

Bases: PrivateDBCredentials

BaseModel for S3 database credentials.

access_key_id: str

credentials_name: str

db_type: Literal[PrivateDatabaseType.S3]

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'access_key_id': FieldInfo(annotation=str, required=True), 'credentials_name': FieldInfo(annotation=str, required=True), 'db_type': FieldInfo(annotation=Literal[<PrivateDatabaseType.S3: 'S3_DB'>], required=True), 'secret_access_key': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

secret_access_key: str

class lomas_core.models.config.Server(*, time_attack: TimeAttack, host_ip: str, host_port: int, log_level: str, reload: bool, workers: int)

Bases: BaseModel

BaseModel for uvicorn server configs.

host_ip: str

host_port: int

log_level: str

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'host_ip': FieldInfo(annotation=str, required=True), 'host_port': FieldInfo(annotation=int, required=True), 'log_level': FieldInfo(annotation=str, required=True), 'reload': FieldInfo(annotation=bool, required=True), 'time_attack': FieldInfo(annotation=TimeAttack, required=True), 'workers': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

reload: bool

time_attack: TimeAttack

workers: int

class lomas_core.models.config.TimeAttack(*, method: TimeAttackMethod, magnitude: float)

Bases: BaseModel

BaseModel for configs to prevent timing attacks.

magnitude: float

method: TimeAttackMethod

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'magnitude': FieldInfo(annotation=float, required=True), 'method': FieldInfo(annotation=TimeAttackMethod, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.config.YamlDBConfig(*, db_type: Literal[AdminDBType.YAML], db_file: str)

Bases: DBConfig

BaseModel for dataset store configs in case of a Yaml database.

db_file: str

db_type: Literal[AdminDBType.YAML]

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'db_file': FieldInfo(annotation=str, required=True), 'db_type': FieldInfo(annotation=Literal[<AdminDBType.YAML: 'yaml'>], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

lomas_core.models.constants module

class lomas_core.models.constants.AdminDBType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

Types of administration databases.

MONGODB = 'mongodb'

YAML = 'yaml'

class lomas_core.models.constants.ConfigKeys(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

Keys of the configuration file.

RUNTIME_ARGS = 'runtime_args'

SETTINGS = 'settings'

class lomas_core.models.constants.ExceptionType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

Lomas server exception types.

To be used as discriminator when parsing corresponding models

EXTERNAL_LIBRARY = 'ExternalLibraryException'

INTERNAL_SERVER = 'InternalServerException'

INVALID_QUERY = 'InvalidQueryException'

UNAUTHORIZED_ACCESS = 'UnauthorizedAccessException'

class lomas_core.models.constants.MetadataColumnType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

Column types for metadata.

BOOLEAN = 'boolean'

CAT_INT = 'categorical_int'

CAT_STRING = 'categorical_string'

DATETIME = 'datetime'

FLOAT = 'float'

INT = 'int'

STRING = 'string'

class lomas_core.models.constants.Precision(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: IntEnum

Precision of integer and float data.

DOUBLE = 64

SINGLE = 32

class lomas_core.models.constants.PrivateDatabaseType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

Type of Private Database for the private data.

PATH = 'PATH_DB'

S3 = 'S3_DB'

class lomas_core.models.constants.TimeAttackMethod(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

Possible methods against timing attacks.

JITTER = 'jitter'

STALL = 'stall'

lomas_core.models.requests module

class lomas_core.models.requests.DiffPrivLibDummyQueryModel(*, dataset_name: str, dummy_nb_rows: Annotated[int, Gt(gt=0)], dummy_seed: int, diffprivlib_json: str, feature_columns: list, target_columns: list | None, test_size: Annotated[float, Gt(gt=0.0), Lt(lt=1.0)], test_train_split_seed: int, imputer_strategy: str)

Bases: DiffPrivLibQueryModel, DummyQueryModel

Input model for a DiffPrivLib query on a dummy dataset.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'dataset_name': 'PENGUIN', 'diffprivlib_json': '{"module": "diffprivlib", "version": "0.6.5", "pipeline": [{"type": "_dpl_type:StandardScaler", "name": "scaler", "params": {"with_mean": true, "with_std": true, "copy": true, "epsilon": 0.5, "bounds": {"_tuple": true, "_items": [[30.0, 13.0, 150.0, 2000.0], [65.0, 23.0, 250.0, 7000.0]]}, "random_state": null, "accountant": "_dpl_instance:BudgetAccountant"}}, {"type": "_dpl_type:LogisticRegression", "name": "classifier", "params": {"tol": 0.0001, "C": 1.0, "fit_intercept": true, "random_state": null, "max_iter": 100, "verbose": 0, "warm_start": false, "n_jobs": null, "epsilon": 1.0, "data_norm": 83.69469642643347, "accountant": "_dpl_instance:BudgetAccountant"}}]}', 'dummy_nb_rows': 100, 'dummy_seed': 42, 'feature_columns': ['bill_length_mm', 'bill_depth_mm', 'flipper_length_mm', 'body_mass_g'], 'imputer_strategy': 'drop', 'target_columns': ['species'], 'test_size': 0.2, 'test_train_split_seed': 4}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'diffprivlib_json': FieldInfo(annotation=str, required=True, description='The DiffPrivLib pipeline for the query (See diffprivlib_logger package.).'), 'dummy_nb_rows': FieldInfo(annotation=int, required=True, description='The number of rows in the dummy dataset.', metadata=[Gt(gt=0)]), 'dummy_seed': FieldInfo(annotation=int, required=True, description='The seed to set at the start of the dummy dataset generation.'), 'feature_columns': FieldInfo(annotation=list, required=True, description='The list of feature columns to train.'), 'imputer_strategy': FieldInfo(annotation=str, required=True, description='The imputation strategy.'), 'target_columns': FieldInfo(annotation=Union[list, NoneType], required=True, description='The list of target columns to predict.'), 'test_size': FieldInfo(annotation=float, required=True, description='The proportion of the test set.', metadata=[Gt(gt=0.0), Lt(lt=1.0)]), 'test_train_split_seed': FieldInfo(annotation=int, required=True, description='The seed for the random train/test split.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.requests.DiffPrivLibQueryModel(*, dataset_name: str, diffprivlib_json: str, feature_columns: list, target_columns: list | None, test_size: Annotated[float, Gt(gt=0.0), Lt(lt=1.0)], test_train_split_seed: int, imputer_strategy: str)

Bases: DiffPrivLibRequestModel, QueryModel

Base input model for a diffprivlib query.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'dataset_name': 'PENGUIN', 'diffprivlib_json': '{"module": "diffprivlib", "version": "0.6.5", "pipeline": [{"type": "_dpl_type:StandardScaler", "name": "scaler", "params": {"with_mean": true, "with_std": true, "copy": true, "epsilon": 0.5, "bounds": {"_tuple": true, "_items": [[30.0, 13.0, 150.0, 2000.0], [65.0, 23.0, 250.0, 7000.0]]}, "random_state": null, "accountant": "_dpl_instance:BudgetAccountant"}}, {"type": "_dpl_type:LogisticRegression", "name": "classifier", "params": {"tol": 0.0001, "C": 1.0, "fit_intercept": true, "random_state": null, "max_iter": 100, "verbose": 0, "warm_start": false, "n_jobs": null, "epsilon": 1.0, "data_norm": 83.69469642643347, "accountant": "_dpl_instance:BudgetAccountant"}}]}', 'feature_columns': ['bill_length_mm', 'bill_depth_mm', 'flipper_length_mm', 'body_mass_g'], 'imputer_strategy': 'drop', 'target_columns': ['species'], 'test_size': 0.2, 'test_train_split_seed': 4}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'diffprivlib_json': FieldInfo(annotation=str, required=True, description='The DiffPrivLib pipeline for the query (See diffprivlib_logger package.).'), 'feature_columns': FieldInfo(annotation=list, required=True, description='The list of feature columns to train.'), 'imputer_strategy': FieldInfo(annotation=str, required=True, description='The imputation strategy.'), 'target_columns': FieldInfo(annotation=Union[list, NoneType], required=True, description='The list of target columns to predict.'), 'test_size': FieldInfo(annotation=float, required=True, description='The proportion of the test set.', metadata=[Gt(gt=0.0), Lt(lt=1.0)]), 'test_train_split_seed': FieldInfo(annotation=int, required=True, description='The seed for the random train/test split.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.requests.DiffPrivLibRequestModel(*, dataset_name: str, diffprivlib_json: str, feature_columns: list, target_columns: list | None, test_size: Annotated[float, Gt(gt=0.0), Lt(lt=1.0)], test_train_split_seed: int, imputer_strategy: str)

Bases: LomasRequestModel

Base input model for a diffprivlib request.

diffprivlib_json: str

The DiffPrivLib pipeline for the query (See diffprivlib_logger package.).

feature_columns: list

The list of feature columns to train.

imputer_strategy: str

The imputation strategy.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'dataset_name': 'PENGUIN', 'diffprivlib_json': '{"module": "diffprivlib", "version": "0.6.5", "pipeline": [{"type": "_dpl_type:StandardScaler", "name": "scaler", "params": {"with_mean": true, "with_std": true, "copy": true, "epsilon": 0.5, "bounds": {"_tuple": true, "_items": [[30.0, 13.0, 150.0, 2000.0], [65.0, 23.0, 250.0, 7000.0]]}, "random_state": null, "accountant": "_dpl_instance:BudgetAccountant"}}, {"type": "_dpl_type:LogisticRegression", "name": "classifier", "params": {"tol": 0.0001, "C": 1.0, "fit_intercept": true, "random_state": null, "max_iter": 100, "verbose": 0, "warm_start": false, "n_jobs": null, "epsilon": 1.0, "data_norm": 83.69469642643347, "accountant": "_dpl_instance:BudgetAccountant"}}]}', 'feature_columns': ['bill_length_mm', 'bill_depth_mm', 'flipper_length_mm', 'body_mass_g'], 'imputer_strategy': 'drop', 'target_columns': ['species'], 'test_size': 0.2, 'test_train_split_seed': 4}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'diffprivlib_json': FieldInfo(annotation=str, required=True, description='The DiffPrivLib pipeline for the query (See diffprivlib_logger package.).'), 'feature_columns': FieldInfo(annotation=list, required=True, description='The list of feature columns to train.'), 'imputer_strategy': FieldInfo(annotation=str, required=True, description='The imputation strategy.'), 'target_columns': FieldInfo(annotation=Union[list, NoneType], required=True, description='The list of target columns to predict.'), 'test_size': FieldInfo(annotation=float, required=True, description='The proportion of the test set.', metadata=[Gt(gt=0.0), Lt(lt=1.0)]), 'test_train_split_seed': FieldInfo(annotation=int, required=True, description='The seed for the random train/test split.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

target_columns: list | None

The list of target columns to predict.

test_size: float

The proportion of the test set.

test_train_split_seed: int

The seed for the random train/test split.

class lomas_core.models.requests.DummyQueryModel(*, dataset_name: str, dummy_nb_rows: Annotated[int, Gt(gt=0)], dummy_seed: int)

Bases: QueryModel

Input model for a query on a dummy dataset.

dummy_nb_rows: int

The number of rows in the dummy dataset.

dummy_seed: int

The seed to set at the start of the dummy dataset generation.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'dummy_nb_rows': FieldInfo(annotation=int, required=True, description='The number of rows in the dummy dataset.', metadata=[Gt(gt=0)]), 'dummy_seed': FieldInfo(annotation=int, required=True, description='The seed to set at the start of the dummy dataset generation.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.requests.GetDummyDataset(*, dataset_name: str, dummy_nb_rows: Annotated[int, Gt(gt=0)], dummy_seed: int)

Bases: LomasRequestModel

Model input to get a dummy dataset.

dummy_nb_rows: int

The number of dummy rows to generate.

dummy_seed: int

The seed for the random generation of the dummy dataset.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'dummy_nb_rows': FieldInfo(annotation=int, required=True, description='The number of dummy rows to generate.', metadata=[Gt(gt=0)]), 'dummy_seed': FieldInfo(annotation=int, required=True, description='The seed for the random generation of the dummy dataset.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.requests.LomasRequestModel(*, dataset_name: str)

Bases: BaseModel

Base class for all types of requests to the lomas server.

We differentiate between requests and queries:

• a request does not necessarily require an algorithm to be executed on the private dataset (e.g. some cost requests).

• a query requires executing an algorithm on a private dataset (or a potentially a dummy).

dataset_name: str

The name of the dataset the request is aimed at.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.requests.OpenDPDummyQueryModel(*, dataset_name: str, dummy_nb_rows: Annotated[int, Gt(gt=0)], dummy_seed: int, opendp_json: str, fixed_delta: Annotated[float | None, Ge(ge=0)])

Bases: OpenDPRequestModel, DummyQueryModel

Input model for an opendp query on a dummy dataset.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'dataset_name': 'PENGUIN', 'dummy_nb_rows': 100, 'dummy_seed': 42, 'fixed_delta': 1e-05, 'opendp_json': '{"version": "0.10.0", "ast": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "constructor", "func": "make_chain_tt", "module": "combinators", "args": [{"_type": "constructor", "func": "make_select_column", "module": "transformations", "kwargs": {"key": "bill_length_mm", "TOA": "String"}}, {"_type": "constructor", "func": "make_split_dataframe", "module": "transformations", "kwargs": {"separator": ",", "col_names": {"_type": "list", "_items": ["species", "island", "bill_length_mm", "bill_depth_mm", "flipper_length_mm", "body_mass_g", "sex"]}}}]}, "rhs": {"_type": "constructor", "func": "then_cast_default", "module": "transformations", "kwargs": {"TOA": "f64"}}}, "rhs": {"_type": "constructor", "func": "then_clamp", "module": "transformations", "kwargs": {"bounds": [30.0, 65.0]}}}, "rhs": {"_type": "constructor", "func": "then_resize", "module": "transformations", "kwargs": {"size": 346, "constant": 43.61}}}, "rhs": {"_type": "constructor", "func": "then_variance", "module": "transformations"}}, "rhs": {"_type": "constructor", "func": "then_laplace", "module": "measurements", "kwargs": {"scale": 5.0}}}}'}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'dummy_nb_rows': FieldInfo(annotation=int, required=True, description='The number of rows in the dummy dataset.', metadata=[Gt(gt=0)]), 'dummy_seed': FieldInfo(annotation=int, required=True, description='The seed to set at the start of the dummy dataset generation.'), 'fixed_delta': FieldInfo(annotation=Union[float, NoneType], required=True, description='If the pipeline measurement is of type "ZeroConcentratedDivergence".\n\n(e.g. with "make_gaussian") then it is converted to "SmoothedMaxDivergence"\nwith "make_zCDP_to_approxDP" (see "opendp measurements documentation at\nhttps://docs.opendp.org/en/stable/api/python/opendp.combinators.html#opendp.combinators.make_zCDP_to_approxDP). # noqa # pylint: disable=C0301\nIn that case a "fixed_delta" must be provided by the user.', metadata=[Ge(ge=0)]), 'opendp_json': FieldInfo(annotation=str, required=True, description='The OpenDP pipeline for the query.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.requests.OpenDPQueryModel(*, dataset_name: str, opendp_json: str, fixed_delta: Annotated[float | None, Ge(ge=0)])

Bases: OpenDPRequestModel, QueryModel

Base input model for an opendp query.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'dataset_name': 'PENGUIN', 'fixed_delta': 1e-05, 'opendp_json': '{"version": "0.10.0", "ast": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "constructor", "func": "make_chain_tt", "module": "combinators", "args": [{"_type": "constructor", "func": "make_select_column", "module": "transformations", "kwargs": {"key": "bill_length_mm", "TOA": "String"}}, {"_type": "constructor", "func": "make_split_dataframe", "module": "transformations", "kwargs": {"separator": ",", "col_names": {"_type": "list", "_items": ["species", "island", "bill_length_mm", "bill_depth_mm", "flipper_length_mm", "body_mass_g", "sex"]}}}]}, "rhs": {"_type": "constructor", "func": "then_cast_default", "module": "transformations", "kwargs": {"TOA": "f64"}}}, "rhs": {"_type": "constructor", "func": "then_clamp", "module": "transformations", "kwargs": {"bounds": [30.0, 65.0]}}}, "rhs": {"_type": "constructor", "func": "then_resize", "module": "transformations", "kwargs": {"size": 346, "constant": 43.61}}}, "rhs": {"_type": "constructor", "func": "then_variance", "module": "transformations"}}, "rhs": {"_type": "constructor", "func": "then_laplace", "module": "measurements", "kwargs": {"scale": 5.0}}}}'}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'fixed_delta': FieldInfo(annotation=Union[float, NoneType], required=True, description='If the pipeline measurement is of type "ZeroConcentratedDivergence".\n\n(e.g. with "make_gaussian") then it is converted to "SmoothedMaxDivergence"\nwith "make_zCDP_to_approxDP" (see "opendp measurements documentation at\nhttps://docs.opendp.org/en/stable/api/python/opendp.combinators.html#opendp.combinators.make_zCDP_to_approxDP). # noqa # pylint: disable=C0301\nIn that case a "fixed_delta" must be provided by the user.', metadata=[Ge(ge=0)]), 'opendp_json': FieldInfo(annotation=str, required=True, description='The OpenDP pipeline for the query.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.requests.OpenDPRequestModel(*, dataset_name: str, opendp_json: str, fixed_delta: Annotated[float | None, Ge(ge=0)])

Bases: LomasRequestModel

Base input model for an opendp request.

fixed_delta: float | None

If the pipeline measurement is of type “ZeroConcentratedDivergence”.

(e.g. with “make_gaussian”) then it is converted to “SmoothedMaxDivergence” with “make_zCDP_to_approxDP” (see “opendp measurements documentation at https://docs.opendp.org/en/stable/api/python/opendp.combinators.html#opendp.combinators.make_zCDP_to_approxDP). # noqa # pylint: disable=C0301 In that case a “fixed_delta” must be provided by the user.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'dataset_name': 'PENGUIN', 'fixed_delta': 1e-05, 'opendp_json': '{"version": "0.10.0", "ast": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "partial_chain", "lhs": {"_type": "constructor", "func": "make_chain_tt", "module": "combinators", "args": [{"_type": "constructor", "func": "make_select_column", "module": "transformations", "kwargs": {"key": "bill_length_mm", "TOA": "String"}}, {"_type": "constructor", "func": "make_split_dataframe", "module": "transformations", "kwargs": {"separator": ",", "col_names": {"_type": "list", "_items": ["species", "island", "bill_length_mm", "bill_depth_mm", "flipper_length_mm", "body_mass_g", "sex"]}}}]}, "rhs": {"_type": "constructor", "func": "then_cast_default", "module": "transformations", "kwargs": {"TOA": "f64"}}}, "rhs": {"_type": "constructor", "func": "then_clamp", "module": "transformations", "kwargs": {"bounds": [30.0, 65.0]}}}, "rhs": {"_type": "constructor", "func": "then_resize", "module": "transformations", "kwargs": {"size": 346, "constant": 43.61}}}, "rhs": {"_type": "constructor", "func": "then_variance", "module": "transformations"}}, "rhs": {"_type": "constructor", "func": "then_laplace", "module": "measurements", "kwargs": {"scale": 5.0}}}}'}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'fixed_delta': FieldInfo(annotation=Union[float, NoneType], required=True, description='If the pipeline measurement is of type "ZeroConcentratedDivergence".\n\n(e.g. with "make_gaussian") then it is converted to "SmoothedMaxDivergence"\nwith "make_zCDP_to_approxDP" (see "opendp measurements documentation at\nhttps://docs.opendp.org/en/stable/api/python/opendp.combinators.html#opendp.combinators.make_zCDP_to_approxDP). # noqa # pylint: disable=C0301\nIn that case a "fixed_delta" must be provided by the user.', metadata=[Ge(ge=0)]), 'opendp_json': FieldInfo(annotation=str, required=True, description='The OpenDP pipeline for the query.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

opendp_json: str

The OpenDP pipeline for the query.

class lomas_core.models.requests.QueryModel(*, dataset_name: str)

Bases: LomasRequestModel

Base input model for any query on a dataset.

We differentiate between requests and queries:

• a request does not necessarily require an algorithm to be executed on the private dataset (e.g. some cost requests).

• a query requires executing an algorithm on a private dataset (or a potentially a dummy).

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.requests.SmartnoiseSQLDummyQueryModel(*, dataset_name: str, dummy_nb_rows: Annotated[int, Gt(gt=0)], dummy_seed: int, query_str: str, epsilon: Annotated[float, Gt(gt=0)], delta: Annotated[float, Ge(ge=0)], mechanisms: dict, postprocess: bool)

Bases: SmartnoiseSQLQueryModel, DummyQueryModel

Input model for a smartnoise-sql query on a dummy dataset.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'dataset_name': 'PENGUIN', 'delta': 1e-05, 'dummy_nb_rows': 100, 'dummy_seed': 42, 'epsilon': 0.1, 'mechanisms': {'count': 'gaussian'}, 'postprocess': True, 'query_str': 'SELECT COUNT(*) AS NB_ROW FROM df'}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'delta': FieldInfo(annotation=float, required=True, description='Privacy parameter (e.g., 1e-5).', metadata=[Ge(ge=0)]), 'dummy_nb_rows': FieldInfo(annotation=int, required=True, description='The number of rows in the dummy dataset.', metadata=[Gt(gt=0)]), 'dummy_seed': FieldInfo(annotation=int, required=True, description='The seed to set at the start of the dummy dataset generation.'), 'epsilon': FieldInfo(annotation=float, required=True, description='Privacy parameter (e.g., 0.1).', metadata=[Gt(gt=0)]), 'mechanisms': FieldInfo(annotation=dict, required=True, description='Dictionary of mechanisms for the query.\n\nSee Smartnoise-SQL mechanisms documentation at\nhttps://docs.smartnoise.org/sql/advanced.html#overriding-mechanisms.'), 'postprocess': FieldInfo(annotation=bool, required=True, description='Whether to postprocess the query results (default: True).\n\nSee Smartnoise-SQL postprocessing documentation\nhttps://docs.smartnoise.org/sql/advanced.html#postprocess.'), 'query_str': FieldInfo(annotation=str, required=True, description='The SQL query to execute.\n\nNOTE: the table name is "df", the query must end with "FROM df"')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.requests.SmartnoiseSQLQueryModel(*, dataset_name: str, query_str: str, epsilon: Annotated[float, Gt(gt=0)], delta: Annotated[float, Ge(ge=0)], mechanisms: dict, postprocess: bool)

Bases: SmartnoiseSQLRequestModel, QueryModel

Base input model for a smartnoise-sql query.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'dataset_name': 'PENGUIN', 'delta': 1e-05, 'epsilon': 0.1, 'mechanisms': {'count': 'gaussian'}, 'postprocess': True, 'query_str': 'SELECT COUNT(*) AS NB_ROW FROM df'}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'delta': FieldInfo(annotation=float, required=True, description='Privacy parameter (e.g., 1e-5).', metadata=[Ge(ge=0)]), 'epsilon': FieldInfo(annotation=float, required=True, description='Privacy parameter (e.g., 0.1).', metadata=[Gt(gt=0)]), 'mechanisms': FieldInfo(annotation=dict, required=True, description='Dictionary of mechanisms for the query.\n\nSee Smartnoise-SQL mechanisms documentation at\nhttps://docs.smartnoise.org/sql/advanced.html#overriding-mechanisms.'), 'postprocess': FieldInfo(annotation=bool, required=True, description='Whether to postprocess the query results (default: True).\n\nSee Smartnoise-SQL postprocessing documentation\nhttps://docs.smartnoise.org/sql/advanced.html#postprocess.'), 'query_str': FieldInfo(annotation=str, required=True, description='The SQL query to execute.\n\nNOTE: the table name is "df", the query must end with "FROM df"')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

postprocess: bool

True).

See Smartnoise-SQL postprocessing documentation https://docs.smartnoise.org/sql/advanced.html#postprocess.

Type:

Whether to postprocess the query results (default

class lomas_core.models.requests.SmartnoiseSQLRequestModel(*, dataset_name: str, query_str: str, epsilon: Annotated[float, Gt(gt=0)], delta: Annotated[float, Ge(ge=0)], mechanisms: dict)

Bases: LomasRequestModel

Base input model for a smarnoise-sql request.

delta: float

Privacy parameter (e.g., 1e-5).

epsilon: float

Privacy parameter (e.g., 0.1).

mechanisms: dict

Dictionary of mechanisms for the query.

See Smartnoise-SQL mechanisms documentation at https://docs.smartnoise.org/sql/advanced.html#overriding-mechanisms.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'dataset_name': 'PENGUIN', 'delta': 1e-05, 'epsilon': 0.1, 'mechanisms': {'count': 'gaussian'}, 'query_str': 'SELECT COUNT(*) AS NB_ROW FROM df'}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'delta': FieldInfo(annotation=float, required=True, description='Privacy parameter (e.g., 1e-5).', metadata=[Ge(ge=0)]), 'epsilon': FieldInfo(annotation=float, required=True, description='Privacy parameter (e.g., 0.1).', metadata=[Gt(gt=0)]), 'mechanisms': FieldInfo(annotation=dict, required=True, description='Dictionary of mechanisms for the query.\n\nSee Smartnoise-SQL mechanisms documentation at\nhttps://docs.smartnoise.org/sql/advanced.html#overriding-mechanisms.'), 'query_str': FieldInfo(annotation=str, required=True, description='The SQL query to execute.\n\nNOTE: the table name is "df", the query must end with "FROM df"')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

query_str: str

The SQL query to execute.

NOTE: the table name is “df”, the query must end with “FROM df”

class lomas_core.models.requests.SmartnoiseSynthDummyQueryModel(*, dataset_name: str, dummy_nb_rows: Annotated[int, Gt(gt=0)], dummy_seed: int, synth_name: SSynthMarginalSynthesizer | SSynthGanSynthesizer, epsilon: Annotated[float, Gt(gt=0)], delta: Annotated[float | None, Ge(ge=0)], select_cols: List, synth_params: dict, nullable: bool, constraints: str, return_model: bool, condition: str, nb_samples: int)

Bases: SmartnoiseSynthQueryModel, DummyQueryModel

Input model for a smarnoise-synth query on a dummy dataset.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'condition': '', 'constraints': '', 'dataset_name': 'PENGUIN', 'delta': 1e-05, 'dummy_nb_rows': 100, 'dummy_seed': 42, 'epsilon': 0.1, 'nb_samples': 200, 'nullable': True, 'return_model': True, 'select_cols': [], 'synth_name': SSynthGanSynthesizer.DP_CTGAN, 'synth_params': {'batch_size': 50, 'embedding_dim': 128, 'epochs': 5}}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'condition': FieldInfo(annotation=str, required=True, description='Sampling condition in `model.sample` (only relevant if return_model is False).'), 'constraints': FieldInfo(annotation=str, required=True, description='Dictionnary for custom table transformer constraints.\n\nColumn that are not specified will be inferred based on metadata.'), 'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'delta': FieldInfo(annotation=Union[float, NoneType], required=True, description='Privacy parameter (e.g., 1e-5).', metadata=[Ge(ge=0)]), 'dummy_nb_rows': FieldInfo(annotation=int, required=True, description='The number of rows in the dummy dataset.', metadata=[Gt(gt=0)]), 'dummy_seed': FieldInfo(annotation=int, required=True, description='The seed to set at the start of the dummy dataset generation.'), 'epsilon': FieldInfo(annotation=float, required=True, description='Privacy parameter (e.g., 0.1).', metadata=[Gt(gt=0)]), 'nb_samples': FieldInfo(annotation=int, required=True, description='Number of samples to generate.\n\n(only relevant if return_model is False)'), 'nullable': FieldInfo(annotation=bool, required=True, description='True if some data cells may be null.'), 'return_model': FieldInfo(annotation=bool, required=True, description='True to get Synthesizer model, False to get samples.'), 'select_cols': FieldInfo(annotation=List, required=True, description='List of columns to select.'), 'synth_name': FieldInfo(annotation=Union[SSynthMarginalSynthesizer, SSynthGanSynthesizer], required=True, description='Name of the synthesizer model to use.'), 'synth_params': FieldInfo(annotation=dict, required=True, description='Keyword arguments to pass to the synthesizer constructor.\n\nSee https://docs.smartnoise.org/synth/synthesizers/index.html#, provide\nall parameters of the model except `epsilon` and `delta`.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.requests.SmartnoiseSynthQueryModel(*, dataset_name: str, synth_name: SSynthMarginalSynthesizer | SSynthGanSynthesizer, epsilon: Annotated[float, Gt(gt=0)], delta: Annotated[float | None, Ge(ge=0)], select_cols: List, synth_params: dict, nullable: bool, constraints: str, return_model: bool, condition: str, nb_samples: int)

Bases: SmartnoiseSynthRequestModel, QueryModel

Base input model for a smarnoise-synth query.

condition: str

Sampling condition in model.sample (only relevant if return_model is False).

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'condition': '', 'constraints': '', 'dataset_name': 'PENGUIN', 'delta': 1e-05, 'epsilon': 0.1, 'nb_samples': 200, 'nullable': True, 'return_model': True, 'select_cols': [], 'synth_name': SSynthGanSynthesizer.DP_CTGAN, 'synth_params': {'batch_size': 50, 'embedding_dim': 128, 'epochs': 5}}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'condition': FieldInfo(annotation=str, required=True, description='Sampling condition in `model.sample` (only relevant if return_model is False).'), 'constraints': FieldInfo(annotation=str, required=True, description='Dictionnary for custom table transformer constraints.\n\nColumn that are not specified will be inferred based on metadata.'), 'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'delta': FieldInfo(annotation=Union[float, NoneType], required=True, description='Privacy parameter (e.g., 1e-5).', metadata=[Ge(ge=0)]), 'epsilon': FieldInfo(annotation=float, required=True, description='Privacy parameter (e.g., 0.1).', metadata=[Gt(gt=0)]), 'nb_samples': FieldInfo(annotation=int, required=True, description='Number of samples to generate.\n\n(only relevant if return_model is False)'), 'nullable': FieldInfo(annotation=bool, required=True, description='True if some data cells may be null.'), 'return_model': FieldInfo(annotation=bool, required=True, description='True to get Synthesizer model, False to get samples.'), 'select_cols': FieldInfo(annotation=List, required=True, description='List of columns to select.'), 'synth_name': FieldInfo(annotation=Union[SSynthMarginalSynthesizer, SSynthGanSynthesizer], required=True, description='Name of the synthesizer model to use.'), 'synth_params': FieldInfo(annotation=dict, required=True, description='Keyword arguments to pass to the synthesizer constructor.\n\nSee https://docs.smartnoise.org/synth/synthesizers/index.html#, provide\nall parameters of the model except `epsilon` and `delta`.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

nb_samples: int

Number of samples to generate.

(only relevant if return_model is False)

return_model: bool

True to get Synthesizer model, False to get samples.

class lomas_core.models.requests.SmartnoiseSynthRequestModel(*, dataset_name: str, synth_name: SSynthMarginalSynthesizer | SSynthGanSynthesizer, epsilon: Annotated[float, Gt(gt=0)], delta: Annotated[float | None, Ge(ge=0)], select_cols: List, synth_params: dict, nullable: bool, constraints: str)

Bases: LomasRequestModel

Base input model for a SmartnoiseSynth request.

constraints: str

Dictionnary for custom table transformer constraints.

Column that are not specified will be inferred based on metadata.

delta: float | None

Privacy parameter (e.g., 1e-5).

epsilon: float

Privacy parameter (e.g., 0.1).

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'json_schema_extra': {'examples': [{'constraints': '', 'dataset_name': 'PENGUIN', 'delta': 1e-05, 'epsilon': 0.1, 'nullable': True, 'select_cols': [], 'synth_name': SSynthGanSynthesizer.DP_CTGAN, 'synth_params': {'batch_size': 50, 'embedding_dim': 128, 'epochs': 5}}]}, 'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'constraints': FieldInfo(annotation=str, required=True, description='Dictionnary for custom table transformer constraints.\n\nColumn that are not specified will be inferred based on metadata.'), 'dataset_name': FieldInfo(annotation=str, required=True, description='The name of the dataset the request is aimed at.'), 'delta': FieldInfo(annotation=Union[float, NoneType], required=True, description='Privacy parameter (e.g., 1e-5).', metadata=[Ge(ge=0)]), 'epsilon': FieldInfo(annotation=float, required=True, description='Privacy parameter (e.g., 0.1).', metadata=[Gt(gt=0)]), 'nullable': FieldInfo(annotation=bool, required=True, description='True if some data cells may be null.'), 'select_cols': FieldInfo(annotation=List, required=True, description='List of columns to select.'), 'synth_name': FieldInfo(annotation=Union[SSynthMarginalSynthesizer, SSynthGanSynthesizer], required=True, description='Name of the synthesizer model to use.'), 'synth_params': FieldInfo(annotation=dict, required=True, description='Keyword arguments to pass to the synthesizer constructor.\n\nSee https://docs.smartnoise.org/synth/synthesizers/index.html#, provide\nall parameters of the model except `epsilon` and `delta`.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

nullable: bool

True if some data cells may be null.

select_cols: List

List of columns to select.

synth_name: SSynthMarginalSynthesizer | SSynthGanSynthesizer

Name of the synthesizer model to use.

synth_params: dict

Keyword arguments to pass to the synthesizer constructor.

See https://docs.smartnoise.org/synth/synthesizers/index.html#, provide all parameters of the model except epsilon and delta.

lomas_core.models.requests.model_input_to_lib(request: LomasRequestModel) → DPLibraries

Return the type of DP library given a LomasRequestModel.

Parameters:

request (LomasRequestModel) – The user request

Raises:

InternalServerException – If the library type cannot be determined.

Returns:

The type of library for the request.

Return type:

DPLibraries

lomas_core.models.responses module

class lomas_core.models.responses.CostResponse(*, epsilon: float, delta: float)

Bases: ResponseModel

Model for responses to cost estimation requests or queries.

delta: float

The delta cost of the query.

epsilon: float

The epsilon cost of the query.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'delta': FieldInfo(annotation=float, required=True, description='The delta cost of the query.'), 'epsilon': FieldInfo(annotation=float, required=True, description='The epsilon cost of the query.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.responses.DiffPrivLibQueryResult(*, res_type: Literal[DPLibraries.DIFFPRIVLIB] = DPLibraries.DIFFPRIVLIB, score: float, model: Annotated[DiffprivlibMixin, PlainSerializer(func=serialize_model, return_type=PydanticUndefined, when_used=always), PlainValidator(func=deserialize_model)])

Bases: BaseModel

Model for diffprivlib query result.

model: Annotated[DiffprivlibMixin, PlainSerializer(func=serialize_model, return_type=PydanticUndefined, when_used=always), PlainValidator(func=deserialize_model)]

The trained model.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'model': FieldInfo(annotation=DiffprivlibMixin, required=True, metadata=[PlainSerializer(func=<function serialize_model>, return_type=PydanticUndefined, when_used='always'), PlainValidator(func=<function deserialize_model>)]), 'res_type': FieldInfo(annotation=Literal[<DPLibraries.DIFFPRIVLIB: 'diffprivlib'>], required=False, default=<DPLibraries.DIFFPRIVLIB: 'diffprivlib'>), 'score': FieldInfo(annotation=float, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

res_type: Literal[DPLibraries.DIFFPRIVLIB]

Result type description.

score: float

The trained model score.

class lomas_core.models.responses.DummyDsResponse(*, dtypes: Dict[str, str], datetime_columns: List[str], dummy_df: Annotated[DataFrame, PlainSerializer(func=dataframe_to_dict, return_type=PydanticUndefined, when_used=always)])

Bases: ResponseModel

Model for responses to dummy dataset requests.

datetime_columns: List[str]

The list of columns with datetime type.

classmethod deserialize_dummy_df(v: DataFrame | dict, info: ValidationInfo) → DataFrame

Decodes the dict representation of the dummy df with correct types.

Only does so if the input value is not already a dataframe. :param v: The dataframe to decode. :type v: pd.DataFrame | dict :param info: Validation info to access other model fields. :type info: ValidationInfo

Returns:

The decoded dataframe.

Return type:

pd.DataFrame

dtypes: Dict[str, str]

The dummy_df column data types.

dummy_df: Annotated[DataFrame, PlainSerializer(func=dataframe_to_dict, return_type=PydanticUndefined, when_used=always)]

The dummy dataframe.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'datetime_columns': FieldInfo(annotation=List[str], required=True), 'dtypes': FieldInfo(annotation=Dict[str, str], required=True), 'dummy_df': FieldInfo(annotation=DataFrame, required=True, metadata=[PlainSerializer(func=<function dataframe_to_dict>, return_type=PydanticUndefined, when_used='always')])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.responses.InitialBudgetResponse(*, initial_epsilon: float, initial_delta: float)

Bases: ResponseModel

Model for responses to initial budget queries.

initial_delta: float

The initial delta privacy loss budget.

initial_epsilon: float

The initial epsilon privacy loss budget.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'initial_delta': FieldInfo(annotation=float, required=True), 'initial_epsilon': FieldInfo(annotation=float, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.responses.OpenDPQueryResult(*, res_type: Literal[DPLibraries.OPENDP] = DPLibraries.OPENDP, value: int | float | List[int | float])

Bases: BaseModel

Type for opendp result.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'res_type': FieldInfo(annotation=Literal[<DPLibraries.OPENDP: 'opendp'>], required=False, default=<DPLibraries.OPENDP: 'opendp'>), 'value': FieldInfo(annotation=Union[int, float, List[Union[int, float]]], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

res_type: Literal[DPLibraries.OPENDP]

Result type description.

value: int | float | List[int | float]

The result value of the query.

class lomas_core.models.responses.QueryResponse(*, epsilon: float, delta: float, requested_by: str, result: Annotated[DiffPrivLibQueryResult | SmartnoiseSQLQueryResult | SmartnoiseSynthModel | SmartnoiseSynthSamples | OpenDPQueryResult, Discriminator(discriminator=res_type, custom_error_type=None, custom_error_message=None, custom_error_context=None)])

Bases: CostResponse

Response to Lomas queries.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'use_attribute_docstrings': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'delta': FieldInfo(annotation=float, required=True, description='The delta cost of the query.'), 'epsilon': FieldInfo(annotation=float, required=True, description='The epsilon cost of the query.'), 'requested_by': FieldInfo(annotation=str, required=True, description='The user that triggered the query.'), 'result': FieldInfo(annotation=Union[DiffPrivLibQueryResult, SmartnoiseSQLQueryResult, SmartnoiseSynthModel, SmartnoiseSynthSamples, OpenDPQueryResult], required=True, description='The query result object.', metadata=[Discriminator(discriminator='res_type', custom_error_type=None, custom_error_message=None, custom_error_context=None)])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

requested_by: str

The user that triggered the query.

result: Annotated[DiffPrivLibQueryResult | SmartnoiseSQLQueryResult | SmartnoiseSynthModel | SmartnoiseSynthSamples | OpenDPQueryResult, Discriminator(discriminator=res_type, custom_error_type=None, custom_error_message=None, custom_error_context=None)]

The query result object.

class lomas_core.models.responses.RemainingBudgetResponse(*, remaining_epsilon: float, remaining_delta: float)

Bases: ResponseModel

Model for responses to remaining budget queries.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'remaining_delta': FieldInfo(annotation=float, required=True), 'remaining_epsilon': FieldInfo(annotation=float, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

remaining_delta: float

The remaining delta privacy loss budget.

remaining_epsilon: float

The remaining epsilon privacy loss budget.

class lomas_core.models.responses.ResponseModel

Bases: BaseModel

Base model for any response from the server.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class lomas_core.models.responses.SmartnoiseSQLQueryResult(*, res_type: Literal[DPLibraries.SMARTNOISE_SQL] = DPLibraries.SMARTNOISE_SQL, df: Annotated[DataFrame, PlainSerializer(func=dataframe_to_dict, return_type=PydanticUndefined, when_used=always), PlainValidator(func=dataframe_from_dict)])

Bases: BaseModel

Type for smartnoise_sql result type.

df: Annotated[DataFrame, PlainSerializer(func=dataframe_to_dict, return_type=PydanticUndefined, when_used=always), PlainValidator(func=dataframe_from_dict)]

Dataframe containing the query result.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'df': FieldInfo(annotation=DataFrame, required=True, metadata=[PlainSerializer(func=<function dataframe_to_dict>, return_type=PydanticUndefined, when_used='always'), PlainValidator(func=<function dataframe_from_dict>)]), 'res_type': FieldInfo(annotation=Literal[<DPLibraries.SMARTNOISE_SQL: 'smartnoise_sql'>], required=False, default=<DPLibraries.SMARTNOISE_SQL: 'smartnoise_sql'>)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

res_type: Literal[DPLibraries.SMARTNOISE_SQL]

Result type description.

class lomas_core.models.responses.SmartnoiseSynthModel(*, res_type: Literal[DPLibraries.SMARTNOISE_SYNTH] = DPLibraries.SMARTNOISE_SYNTH, model: Annotated[Synthesizer, PlainSerializer(func=serialize_model, return_type=PydanticUndefined, when_used=always), PlainValidator(func=deserialize_model)])

Bases: BaseModel

Type for smartnoise_synth result when it is a pickled model.

model: Annotated[Synthesizer, PlainSerializer(func=serialize_model, return_type=PydanticUndefined, when_used=always), PlainValidator(func=deserialize_model)]

Synthetic data generator model.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'model': FieldInfo(annotation=Synthesizer, required=True, metadata=[PlainSerializer(func=<function serialize_model>, return_type=PydanticUndefined, when_used='always'), PlainValidator(func=<function deserialize_model>)]), 'res_type': FieldInfo(annotation=Literal[<DPLibraries.SMARTNOISE_SYNTH: 'smartnoise_synth'>], required=False, default=<DPLibraries.SMARTNOISE_SYNTH: 'smartnoise_synth'>)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

res_type: Literal[DPLibraries.SMARTNOISE_SYNTH]

Result type description.

class lomas_core.models.responses.SmartnoiseSynthSamples(*, res_type: Literal['sn_synth_samples'] = 'sn_synth_samples', df_samples: Annotated[DataFrame, PlainSerializer(func=dataframe_to_dict, return_type=PydanticUndefined, when_used=always), PlainValidator(func=dataframe_from_dict)])

Bases: BaseModel

Type for smartnoise_synth result when it is a dataframe of samples.

df_samples: Annotated[DataFrame, PlainSerializer(func=dataframe_to_dict, return_type=PydanticUndefined, when_used=always), PlainValidator(func=dataframe_from_dict)]

Dataframe containing the generated synthetic samples.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'df_samples': FieldInfo(annotation=DataFrame, required=True, metadata=[PlainSerializer(func=<function dataframe_to_dict>, return_type=PydanticUndefined, when_used='always'), PlainValidator(func=<function dataframe_from_dict>)]), 'res_type': FieldInfo(annotation=Literal['sn_synth_samples'], required=False, default='sn_synth_samples')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

res_type: Literal['sn_synth_samples']

Result type description.

class lomas_core.models.responses.SpentBudgetResponse(*, total_spent_epsilon: float, total_spent_delta: float)

Bases: ResponseModel

Model for responses to spent budget queries.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'total_spent_delta': FieldInfo(annotation=float, required=True), 'total_spent_epsilon': FieldInfo(annotation=float, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

total_spent_delta: float

The total spent delta privacy loss budget.

total_spent_epsilon: float

The total spent epsilon privacy loss budget.

lomas_core.models.utils module

lomas_core.models.utils.dataframe_from_dict(serialized_df: DataFrame | dict) → DataFrame

Transforms input dict into pandas dataframe.

If the input is already a dataframe, it is simply returned unmodified.

Parameters:

serialized_df (pd.DataFrame | dict) – Dataframe in dict format. Or pd.Dataframe.

Returns:

The transformed dataframe.

Return type:

pd.DataFrame

lomas_core.models.utils.dataframe_to_dict(df: DataFrame) → dict

Transforms pandas dataframe into a dictionary.

Parameters:

df (pd.DataFrame) – The dataframe to “serialize”.

Returns:

The pandas dataframe in dictionary format.

Return type:

dict

lomas_core.models.utils.deserialize_model(serialized_model: Any) → Any

Deserialize a base64 encoded byte string into a python object.

Parameters:

serialized_model (Any) – Encoded python object.

Returns:

Deserialized python object.

Return type:

Any

lomas_core.models.utils.serialize_model(model: Any) → str

Serialise a python object into an utf-8 string.

Fitted Smartnoise Synth synthesizer or fitted DiffPrivLib pipeline.

Parameters:

model (Any) – An object to serialise

Returns:

string of serialised model

Return type:

str

Module contents