Data model¶
Data may come from two distinct sources in pglift: either as objects built by inspecting the system or as manifest documents obtained from the user through an interface. The former kind of model is referred to as the system model whereas the later is called the interface model.
System model¶
- class pglift.models.PostgreSQLInstance(name: str, version, settings: Settings)¶
A bare PostgreSQL instance.
- classmethod creating(name: str, version: str, settings: Settings) Iterator[Self] ¶
Context manager to initialize a PostgreSQLInstance while deferring its “checks” at exit thus leaving the opportunity for actual creation and configuration tasks to run within the context.
- classmethod system_lookup(name: str, version: str, settings: Settings) Self ¶
Build a PostgreSQLInstance by system lookup.
- Raises:
InstanceNotFound – if the instance could not be found by system lookup.
- classmethod from_qualname(value: str, settings: Settings) Self ¶
Lookup for an Instance by its qualified name.
- configuration(managed_only: bool = False) Configuration ¶
Read and return instance configuration.
- class pglift.models.Instance(postgresql: PostgreSQLInstance, services: list[Any] = NOTHING)¶
A PostgreSQL instance with satellite services.
- service(stype: type[S]) S ¶
Return bound satellite service object matching requested type.
- Raises:
ValueError – if not found.
Interface model¶
- class pglift.models.interface.InstanceListItem(*, name: str, version: str, port: int, datadir: Path, status: str)¶
- model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True, 'validate_assignment': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class pglift.models.interface.Instance(*, name: Annotated[str, _PydanticGeneralMetadata(pattern='^[^/-]+$')], version: Annotated[PostgreSQLVersion, BeforeValidator(func=_default_version, json_schema_input_type=PydanticUndefined)] = None, standby: Standby | None = None, upgrading_from: Annotated[PostgreSQLInstanceRef | None, Hidden(), Hidden()] = None, port: Annotated[Annotated[int, AfterValidator(func=check_port_available)] | None, AfterValidator(func=_port_unset_is_available)] = None, settings: Annotated[MutableMapping[str, Any], AfterValidator(func=_no_port_in_settings), Hidden()] = {}, data_checksums: bool | None = None, locale: str | None = None, encoding: str | None = None, auth: Auth | None = None, surole_password: Annotated[SecretStr | None, Option(name=None, metavar='password'), AfterValidator(func=_password_required_for_local_auth)] = None, replrole_password: Annotated[SecretStr | None, Option(name=None, metavar='password')] = None, pending_restart: Annotated[bool, Hidden(), Hidden()] = False, state: Annotated[Literal['stopped', 'started', 'absent', 'restarted'], Choices(name=None, choices=['started', 'stopped'])] = 'started', creating: Annotated[bool, Hidden(), Hidden(), AfterValidator(func=_set_creating_when_upgrading_from)] = False, databases: Annotated[list[Database], Hidden()] = [], roles: Annotated[list[Role], Hidden()] = [], replication_slots: _replication_slots_not_with_demoted_standby)] = [], restart_on_changes: Annotated[bool, Hidden()] = False, **extra_data: Any)¶
A pglift instance, on top of a PostgreSQL instance.
This combines the definition of a base PostgreSQL instance with extra satellite components or cluster objects.
When unspecified, some fields values are computed from site settings and site templates, the combination of which serves as a default “template” for the Instance model.
- service(stype: type[_S]) _S ¶
Return satellite Service attached to this instance.
- Raises:
ValueError – if not found.
- model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'frozen': True, 'validate_assignment': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class pglift.models.interface.Role(*, name: str, state: ~typing.Annotated[~typing.Literal['present', 'absent'], Hidden()] = 'present', password: ~pydantic.types.SecretStr | None = None, encrypted_password: ~typing.Annotated[~pydantic.types.SecretStr | None, ~pydantic.functional_validators.AfterValidator(func=functools.partial(<function check_mutually_exclusive_with at 0x7309d448dda0>, 'password'))] = None, drop_owned: ~typing.Annotated[bool, Hidden(), ~pydantic.functional_validators.AfterValidator(func=~pglift.models.interface.validate_state_is_absent)] = False, reassign_owned: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1), Hidden(), ~pydantic.functional_validators.AfterValidator(func=~pglift.models.interface.validate_state_is_absent)] = None, has_password: ~typing.Annotated[bool, Hidden(), Hidden(), ~pydantic.functional_validators.AfterValidator(func=~pglift.models.interface._set_has_password)] = False, inherit: bool = True, login: bool = False, superuser: bool = False, createdb: bool = False, createrole: bool = False, replication: bool = False, connection_limit: int | None = None, valid_until: ~datetime.datetime | None = None, validity: ~typing.Annotated[~datetime.datetime | None, ~pydantic.functional_validators.AfterValidator(func=functools.partial(<function check_mutually_exclusive_with at 0x7309d448dda0>, 'valid_until', operations={'create', 'update'})), ~pydantic.functional_validators.AfterValidator(func=~pglift.models.interface._validate_role_validity)] = None, memberships: ~typing.Annotated[list[~typing.Annotated[~pglift.models.interface.RoleMembership, ~pydantic.functional_validators.BeforeValidator(func=functools.partial(<function as_dict at 0x7309d43d2520>, key='role'), json_schema_input_type=PydanticUndefined), ~pydantic.functional_serializers.WrapSerializer(func=functools.partial(<function serialize at 0x7309d42ac0e0>, key='role'), return_type=PydanticUndefined, when_used=always)]], ListOption(name='in_role', metavar='role', item_key='role', names={'add': '--grant', 'remove': '--revoke'}, descriptions={'add': 'Grant membership of the given role.', 'remove': 'Revoke membership of the given role.'})] = [], hba_records: ~typing.Annotated[list[~pglift.models.interface.HbaRecordForRole], Hidden()] = [], **extra_data: ~typing.Any)¶
PostgreSQL role
- model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'frozen': True, 'validate_assignment': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class pglift.models.interface.Database(*, name: str, force_drop: Annotated[bool, Hidden(), AfterValidator(func=validate_state_is_absent)] = False, state: Annotated[Literal['present', 'absent'], Hidden()] = 'present', owner: str | None = None, settings: timedelta | None] | None, Hidden(), Spec(spec={'type': 'dict', 'required': False})] = None, schemas: _set_schemas_owner)] = [], extensions: '})] = [], locale: str | None = None, publications: Annotated[list[Publication], Hidden()] = [], subscriptions: Annotated[list[Subscription], Hidden()] = [], clone: CloneOptions | None = None, tablespace: Annotated[str | None, AfterValidator(func=check_tablespace)] = None)¶
PostgreSQL database
- model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True, 'validate_assignment': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class pglift.models.interface.Tablespace(*, name: str, location: str, size: ~typing.Annotated[int, <pglift.types.ByteSizeType object at 0x7309d49796a0>])¶
- model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True, 'validate_assignment': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class pglift.models.interface.DefaultPrivilege(*, database: str, schema: str, object_type: str, role: str, privileges: Annotated[list[str], AfterValidator(func=_sort)])¶
Default access privilege
- model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True, 'validate_assignment': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class pglift.models.interface.Privilege(*, database: str, schema: str, object_type: str, role: str, privileges: Annotated[list[str], AfterValidator(func=_sort)], object_name: str, column_privileges: Annotated[Mapping[str, list[str]], AfterValidator(func=_sort_values)])¶
Access privilege
- model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True, 'validate_assignment': True}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
Results¶
- class pglift.models.interface.ApplyResult(*, change_state: ~typing.Literal['created', 'changed', 'dropped'] | None = None, diff: ~typing.Any | None = <factory>)¶
ApplyResult allows to describe the result of a call to apply function (Eg: pglift.database.apply) to an object (Eg: database, instance,…).
- The change_state attribute of this class can be set to one of to those values:
‘created’ if the object has been created,
‘changed’ if the object has been changed,
‘dropped’ if the object has been dropped,
None
if nothing happened to the object we manipulate (neither created, changed or dropped)