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.

property qualname: str¶: Version qualified name, e.g. 13-main.

configuration(managed_only: bool = False) → Configuration¶: Read and return instance configuration.

property port: int¶: TCP port the server listens on.

property socket_directory: str | None¶

Directory path in which the socket should be.

This is determined from ‘unix_socket_directories’ configuration entry, only considering the first item not starting with @. None if that setting is not defined.

property dumps_directory: Path¶: Path to directory where database dumps are stored.

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.

class pglift.models.PGSetting(name: str, setting: str, context: str, pending_restart: bool)¶: A column from pg_settings view.

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].

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.

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)