policyweaver.models.config¶

class policyweaver.models.config.CatalogItem(*, catalog: str | None = None, catalog_schema: str | None = None, table: str | None = None)¶

Bases: CommonBaseModel

Base model for catalog items. This model provides common functionality for catalog items such as ID, name, and type. Attributes:

id (str): The unique identifier for the catalog item. name (str): The name of the catalog item. type (str): The type of the catalog item. catalog (str): The catalog to which the item belongs. catalog_schema (str): The schema of the catalog. table (str): The table associated with the catalog item.

_abc_impl = <_abc._abc_data object>¶

catalog: str | None¶

catalog_schema: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

table: str | None¶

class policyweaver.models.config.ColumnConstraintsConfig(*, columnlevelsecurity: bool | None = False, fallback: str | None = 'deny')¶

Bases: CommonBaseModel

Configuration for column constraints in the fabric. Attributes:

columnlevelsecurity (bool): Flag to indicate whether column-level security is enabled. fallback (str): The fallback policy, e.g., “deny” or “allow”.

_abc_impl = <_abc._abc_data object>¶

columnlevelsecurity: bool | None¶

fallback: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class policyweaver.models.config.ConstraintsConfig(*, columns: ColumnConstraintsConfig | None = None, rows: RowConstraintsConfig | None = None)¶

Bases: CommonBaseModel

Configuration for fabric constraints in the Policy Weaver application. Attributes:

columns (dict): A dictionary containing column-level security settings.

columnlevelsecurity (bool): Flag to indicate whether column-level security is enabled.

fallback (str): The fallback policy, e.g., “deny” or “allow”.

_abc_impl = <_abc._abc_data object>¶

columns: ColumnConstraintsConfig | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rows: RowConstraintsConfig | None¶

Bases: CommonBaseModel

Configuration for the fabric in the Policy Weaver application. Attributes:

tenant_id (str): The Azure tenant ID. workspace_id (str): The Azure workspace ID. workspace_name (str): The name of the Azure workspace. mirror_id (str): The ID of the mirror in the fabric. mirror_name (str): The name of the mirror in the fabric. fabric_role_suffix (str): The suffix for the fabric role, default is “PWPolicy”. delete_default_reader_role (bool): Flag to indicate whether to delete the default reader role,

default is False.

_abc_impl = <_abc._abc_data object>¶

delete_default_reader_role: bool | None¶

fabric_role_suffix: str | None¶

mirror_id: str | None¶

mirror_name: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

policy_mapping: str | None¶

tenant_id: str | None¶

workspace_id: str | None¶

workspace_name: str | None¶

class policyweaver.models.config.KeyVaultConfig(*, use_key_vault: bool | None = False, name: str | None = None, authentication_method: str | None = None)¶

Bases: CommonBaseModel

Configuration for Azure Key Vault. Attributes:

use_key_vault (bool): Flag to indicate whether to use Azure Key Vault. name (str): The name of the Key Vault. authentication_method (str): The authentication method to use for accessing the Key Vault.

_abc_impl = <_abc._abc_data object>¶

authentication_method: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None¶

use_key_vault: bool | None¶

class policyweaver.models.config.RowConstraintsConfig(*, rowlevelsecurity: bool | None = False, fallback: str | None = 'deny')¶

Bases: CommonBaseModel

Configuration for row constraints in the fabric. Attributes:

rowlevelsecurity (bool): Flag to indicate whether row-level security is enabled. fallback (str): The fallback policy, e.g., “deny” or “allow”.

_abc_impl = <_abc._abc_data object>¶

fallback: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rowlevelsecurity: bool | None¶

class policyweaver.models.config.ServicePrincipalConfig(*, tenant_id: str | None = None, client_id: str | None = None, client_secret: str | None = None)¶

Bases: CommonBaseModel

Configuration for service principal authentication. Attributes:

tenant_id (str): The Azure tenant ID. client_id (str): The client ID of the service principal. client_secret (str): The client secret of the service principal.

_abc_impl = <_abc._abc_data object>¶

client_id: str | None¶

client_secret: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tenant_id: str | None¶

class policyweaver.models.config.Source(*, name: str | None = None, schemas: List[SourceSchema] | None = None)¶

Bases: CommonBaseModel

Represents a source in the Policy Weaver application. Attributes:

id (str): The unique identifier for the source. type (PolicyWeaverConnectorType): The type of the source connector. name (str): The name of the source. schemas (List[SourceSchema]): A list of schemas in the source.

_abc_impl = <_abc._abc_data object>¶

get_schema_list() → List[str]¶: Returns a list of schema names in the source. If there are no schemas, returns None. Returns:

List[str]: A list of schema names if schemas exist, otherwise None.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None¶

schemas: List[SourceSchema] | None¶

class policyweaver.models.config.SourceMap(*, application_name: str | None = 'POLICY_WEAVER', correlation_id: str | None = None, type: PolicyWeaverConnectorType | None = None, source: Source | None = None, fabric: FabricConfig | None = None, constraints: ConstraintsConfig | None = None, service_principal: ServicePrincipalConfig | None = None, mapped_items: List[SourceMapItem] | None = None, keyvault: KeyVaultConfig | None = None)¶

Bases: CommonBaseModel

Represents a source map in the Policy Weaver application. This model contains configuration for the source map, including application name, correlation ID, connector type, source, fabric configuration, service principal configuration, and mapped items. Attributes:

application_name (str): The name of the application using the source map. correlation_id (str): A unique identifier for the correlation of operations. type (PolicyWeaverConnectorType): The type of the connector used in the source map. source (Source): The source from which the policies are mapped. fabric (FabricConfig): Configuration for the fabric in which the policies are managed. service_principal (ServicePrincipalConfig): Configuration for service principal authentication. mapped_items (List[SourceMapItem]): A list of items that are mapped in the source map.

_abc_impl = <_abc._abc_data object>¶

application_name: str | None¶

constraints: ConstraintsConfig | None¶

correlation_id: str | None¶

fabric: FabricConfig | None¶

classmethod from_yaml(path: str = None) → SourceMap¶

Load a SourceMap instance from a YAML file. Args:

path (str): The path to the YAML file. If None, uses the default paths.

Returns:: SourceMap: An instance of SourceMap loaded from the YAML file.
Raises:: PolicyWeaverError: If the YAML file does not exist or cannot be loaded.

keyvault: KeyVaultConfig | None¶

mapped_items: List[SourceMapItem] | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None¶

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:: self: The BaseModel instance. context: The context.

service_principal: ServicePrincipalConfig | None¶

source: Source | None¶

to_yaml(path: str = None) → None¶

Save the SourceMap instance to a YAML file. Args:

path (str): The path to save the YAML file. If None, uses the default paths.

Raises:: IOError: If none of the paths are writable.

type: PolicyWeaverConnectorType | None¶

class policyweaver.models.config.SourceMapItem(*, catalog: str | None = None, catalog_schema: str | None = None, table: str | None = None, mirror_table_name: str | None = None)¶

Bases: CatalogItem

Represents an item in the source map. Attributes:

id (str): The unique identifier for the source map item. name (str): The name of the source map item. type (str): The type of the source map item. catalog (str): The catalog to which the source map item belongs. catalog_schema (str): The schema of the catalog. table (str): The table associated with the source map item. mirror_table_name (str): The name of the mirror table associated with the source map item.

_abc_impl = <_abc._abc_data object>¶

mirror_table_name: str | None¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class policyweaver.models.config.SourceSchema(*, name: str | None = None, tables: List[str] | None = None)¶

Bases: CommonBaseModel

Represents a schema in a source. Attributes:

name (str): The name of the schema. tables (List[str]): A list of table names in the schema.

_abc_impl = <_abc._abc_data object>¶

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'exclude_none': True, 'exclude_unset': True, 'populate_by_name': True, 'use_enum_values': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None¶

tables: List[str] | None¶