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, version, settings)

A bare PostgreSQL instance.

Parameters:

  • name (str)

  • version (Literal['18', '17', '16', '15', '14'])

  • settings (Settings)

classmethod creating(name, version, settings)

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.

Parameters:

  • name (str)

  • version (Literal['18', '17', '16', '15', '14'])

  • settings (Settings)

Return type:

Iterator[Self]

classmethod system_lookup(name, version, settings)

Build a PostgreSQLInstance by system lookup.

Raises:

InstanceNotFound – if the instance could not be found by system lookup.

Parameters:

  • name (str)

  • version (Literal['18', '17', '16', '15', '14'])

  • settings (Settings)

Return type:

Self

classmethod from_qualname(value, settings)

Lookup for an Instance by its qualified name.

Parameters:

  • value (str)

  • settings (Settings)

Return type:

Self

property qualname: str

Version qualified name, e.g. 18-main.

configuration(managed_only=False)

Read and return instance configuration.

Parameters:

managed_only (bool)

Return type:

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, services=NOTHING)

A PostgreSQL instance with satellite services.

Parameters:
service(stype)

Return bound satellite service object matching requested type.

Raises:

ValueError – if not found.

Parameters:

stype (type[S])

Return type:

S

class pglift.models.PGSetting(name, setting, context, pending_restart)

A column from pg_settings view.

Parameters:

Interface model

class pglift.models.interface.InstanceListItem(*, name, version, port, datadir, status)
Parameters:

  • name (Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Instance name.')])

  • version (Annotated[Literal['18', '17', '16', '15', '14'], FieldInfo(annotation=NoneType, required=True, description='PostgreSQL version.')])

  • port (Annotated[int, FieldInfo(annotation=NoneType, required=True, description='TCP port the PostgreSQL instance is listening to.')])

  • datadir (Annotated[Path, FieldInfo(annotation=NoneType, required=True, description='PostgreSQL data directory.')])

  • status (Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Runtime status.')])

class pglift.models.interface.Instance(*, name, version=None, standby=None, upgrading_from=None, port=None, settings={}, data_checksums=None, locale=None, encoding=None, auth=None, surole_password=None, replrole_password=None, pending_restart=False, state='started', creating=False, databases=[], roles=[], replication_slots=[], restart_on_changes=False, **extra_data)

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.

Parameters:

  • name (Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Instance name.', json_schema_extra={'readOnly': True}, metadata=[_PydanticGeneralMetadata(pattern='^[^/-]+$')])])

  • version (Annotated[Literal['18', '17', '16', '15', '14'], FieldInfo(annotation=NoneType, required=False, default=None, description='PostgreSQL version; if unspecified, determined from site settings or most recent PostgreSQL installation available on site.', json_schema_extra={'readOnly': True}, validate_default=True), ~pydantic.functional_validators.BeforeValidator(func=~pglift.models.interface._default_version, json_schema_input_type=PydanticUndefined)])

  • standby (Annotated[Standby | None, FieldInfo(annotation=NoneType, required=True, description='Standby information.')])

  • upgrading_from (Annotated[PostgreSQLInstanceRef | None, Hidden(), Hidden(), FieldInfo(annotation=NoneType, required=True, description='Internal field to keep a reference to the instance this manifest will be upgraded from.', exclude=True)])

  • port (Annotated[Annotated[int, AfterValidator(func=~pglift.models.validators.check_port_available)] | None, AfterValidator(func=~pglift.models.interface._port_unset_is_available), FieldInfo(annotation=NoneType, required=True, description='TCP port the PostgreSQL instance will be listening to.', validate_default=True)])

  • settings (Annotated[MutableMapping[str, Any], AfterValidator(func=~pglift.models.interface._no_port_in_settings), Hidden(), FieldInfo(annotation=NoneType, required=True, description='Settings for the PostgreSQL instance.', json_schema_extra={'examples': [{'listen_addresses': '*', 'shared_buffers': '1GB', 'ssl': True, 'ssl_key_file': '/etc/certs/db.key', 'ssl_cert_file': '/etc/certs/db.key', 'shared_preload_libraries': 'pg_stat_statements'}]})])

  • data_checksums (Annotated[bool | None, FieldInfo(annotation=NoneType, required=True, description='Enable or disable data checksums. If unspecified, fall back to site settings choice.')])

  • locale (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Default locale.', json_schema_extra={'readOnly': True})])

  • encoding (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Character encoding of the PostgreSQL instance.', json_schema_extra={'readOnly': True})])

  • auth (Annotated[Auth | None, FieldInfo(annotation=NoneType, required=True, exclude=True, json_schema_extra={'writeOnly': True})])

  • surole_password (Annotated[SecretStr | None, Option(name=None, metavar='password'), AfterValidator(func=~pglift.models.interface._password_required_for_local_auth), FieldInfo(annotation=NoneType, required=True, description='Super-user role password.', exclude=True, json_schema_extra={'readOnly': True}, validate_default=True)])

  • replrole_password (Annotated[SecretStr | None, Option(name=None, metavar='password'), FieldInfo(annotation=NoneType, required=True, description='Replication role password.', exclude=True, json_schema_extra={'readOnly': True})])

  • pending_restart (Annotated[bool, Hidden(), Hidden(), FieldInfo(annotation=NoneType, required=True, description='Whether the instance needs a restart to account for settings changes.', json_schema_extra={'readOnly': True})])

  • state (Annotated[Literal['stopped', 'started', 'absent', 'restarted'], Choices(name=None, choices=['started', 'stopped']), FieldInfo(annotation=NoneType, required=True, description='Runtime state.')])

  • creating (Annotated[bool, Hidden(), Hidden(), AfterValidator(func=~pglift.models.interface._set_creating_when_upgrading_from), FieldInfo(annotation=NoneType, required=True, description='Internal field to indicate that the instance is being created.', exclude=True, validate_default=True)])

  • databases (Annotated[list[Database], Hidden(), FieldInfo(annotation=NoneType, required=True, description='Databases defined in this instance (non-exhaustive list).', exclude=True, json_schema_extra={'writeOnly': True})])

  • roles (Annotated[list[Role], Hidden(), FieldInfo(annotation=NoneType, required=True, description='Roles defined in this instance (non-exhaustive list).', exclude=True, json_schema_extra={'writeOnly': True})])

  • replication_slots (Annotated[list[Annotated[ReplicationSlot, BeforeValidator(func=~pglift.models.interface.as_dict, json_schema_input_type=PydanticUndefined), WrapSerializer(func=~pglift.models.interface.serialize, return_type=PydanticUndefined, when_used=always)]], ListOption(name='slot', metavar=None, item_key='name', names={'add': '--create-slot', 'remove': '--drop-slot'}, descriptions={'add': 'Replication slots to create in this instance.', 'remove': 'Replication slots to drop from this instance'}), FieldInfo(annotation=NoneType, required=True, description='Replication slots in this instance (non-exhaustive list).', json_schema_extra={'examples': [{'name': 'myslot'}, 'someslot']}), AfterValidator(func=~pglift.models.interface._replication_slots_not_with_demoted_standby)])

  • restart_on_changes (Annotated[bool, Hidden(), FieldInfo(annotation=NoneType, required=True, description='Whether or not to automatically restart the instance to account for settings changes.', exclude=True, json_schema_extra={'writeOnly': True})])

  • extra_data (Any)

service(stype)

Return satellite Service attached to this instance.

Raises:

ValueError – if not found.

Parameters:

stype (type[_S])

Return type:

_S

model_post_init(context, /)

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.

Parameters:

  • self (BaseModel)

  • context (Any)

Return type:

None

class pglift.models.interface.Role(*, name, state='present', password=None, encrypted_password=None, drop_owned=False, reassign_owned=None, has_password=False, inherit=True, login=False, superuser=False, createdb=False, createrole=False, replication=False, connection_limit=None, valid_until=None, memberships=[], hba_records=[], **extra_data)

PostgreSQL role

Parameters:

  • name (Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Role name.', json_schema_extra={'readOnly': True})])

  • state (Annotated[Literal['present', 'absent'], Hidden(), FieldInfo(annotation=NoneType, required=True, description='Whether the role be present or absent.', exclude=True)])

  • password (Literal[None])

  • encrypted_password (Literal[None])

  • drop_owned (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description="Drop all PostgreSQL's objects owned by the role being dropped.", exclude=True), Hidden(), AfterValidator(func=~pglift.models.interface.validate_state_is_absent)])

  • reassign_owned (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description="Reassign all PostgreSQL's objects owned by the role being dropped to the specified role name.", exclude=True, metadata=[MinLen(min_length=1)]), Hidden(), AfterValidator(func=~pglift.models.interface.validate_state_is_absent)])

  • has_password (Annotated[bool, Hidden(), Hidden(), FieldInfo(annotation=NoneType, required=True, description='True if the role has a password.', json_schema_extra={'readOnly': True}, validate_default=True), AfterValidator(func=~pglift.models.interface._set_has_password)])

  • inherit (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Let the role inherit the privileges of the roles it is a member of.')])

  • login (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Allow the role to log in.')])

  • superuser (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether the role is a superuser.')])

  • createdb (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether role can create new databases.')])

  • createrole (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether role can create new roles.')])

  • replication (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Whether the role is a replication role.')])

  • connection_limit (Annotated[int | None, FieldInfo(annotation=NoneType, required=True, description='How many concurrent connections the role can make.')])

  • valid_until (Annotated[datetime | None, FieldInfo(annotation=NoneType, required=True, description="Date and time after which the role's password is no longer valid.")])

  • memberships (Annotated[list[Annotated[RoleMembership, BeforeValidator(func=functools.partial(<function as_dict at 0x7ba076268040>, key='role'), json_schema_input_type=PydanticUndefined), WrapSerializer(func=functools.partial(<function serialize at 0x7ba076164d50>, 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.'}), FieldInfo(annotation=NoneType, required=True, description='Roles which this role should be a member of.')])

  • hba_records (Annotated[list[HbaRecordForRole], Hidden(), FieldInfo(annotation=NoneType, required=True, description='Entries in the pg_hba.conf file for this role.')])

  • extra_data (Any)

class pglift.models.interface.Database(*, name, force_drop=False, state='present', owner=None, settings=None, schemas=[], extensions=[], locale=None, publications=[], subscriptions=[], clone=None, tablespace=None)

PostgreSQL database

Parameters:

  • name (Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Database name.', json_schema_extra={'readOnly': True, 'examples': ['demo']})])

  • force_drop (Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Force the drop.', exclude=True), Hidden(), AfterValidator(func=~pglift.models.interface.validate_state_is_absent)])

  • state (Annotated[Literal['present', 'absent'], Hidden(), FieldInfo(annotation=NoneType, required=True, description='Database state.', exclude=True, json_schema_extra={'examples': ['present']})])

  • owner (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='The role name of the user who will own the database.', json_schema_extra={'examples': ['postgres']})])

  • settings (Annotated[MutableMapping[str, str | bool | float | int | timedelta | None] | None, Hidden(), Spec(spec={'type': 'dict', 'required': False}), FieldInfo(annotation=NoneType, required=True, description='Session defaults for run-time configuration variables for the database. Upon update, an empty (dict) value would reset all settings.', json_schema_extra={'examples': [{'work_mem': '5MB'}]})])

  • schemas (Annotated[list[Annotated[Schema, BeforeValidator(func=~pglift.models.interface.as_dict, json_schema_input_type=PydanticUndefined), WrapSerializer(func=~pglift.models.interface.serialize, return_type=PydanticUndefined, when_used=always)]], ListOption(name='schema', metavar=None, item_key='name', names=None, descriptions={'add': 'Schemas to add to this database.', 'remove': 'Schemas to remove from this database.'}), FieldInfo(annotation=NoneType, required=True, description='Schemas in this database.', json_schema_extra={'examples': [{'name': 'sales'}, 'accounting']}), AfterValidator(func=~pglift.models.interface._set_schemas_owner)])

  • extensions (Annotated[list[Annotated[Extension, BeforeValidator(func=~pglift.models.interface.as_dict, json_schema_input_type=PydanticUndefined), WrapSerializer(func=~pglift.models.interface.serialize, return_type=PydanticUndefined, when_used=always)]], ListOption(name='extension', metavar=None, item_key='name', names=None, descriptions={'add': 'Extensions to add to this database.', 'remove': 'Extensions to remove from this database.'}), FieldInfo(annotation=NoneType, required=True, description='Extensions in this database.', json_schema_extra={'examples': [{'name': 'unaccent', 'schema': 'ext', 'version': '1.0'}, 'hstore']})])

  • locale (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='Locale for this database. Database will be created from template0 if the locale differs from the one set for template1.', json_schema_extra={'examples': ['fr_FR.utf8'], 'readOnly': True, 'writeOnly': True})])

  • publications (Annotated[list[Publication], Hidden(), FieldInfo(annotation=NoneType, required=True, description='Publications in this database.', json_schema_extra={'examples': [{'name': 'mypub'}]})])

  • subscriptions (Annotated[list[Subscription], Hidden(), FieldInfo(annotation=NoneType, required=True, description='Subscriptions in this database.', json_schema_extra={'examples': [{'name': 'mysub', 'publications': ['mypub'], 'enabled': False}]})])

  • clone (Annotated[CloneOptions | None, FieldInfo(annotation=NoneType, required=True, description='Options for cloning a database into this one.', exclude=True, json_schema_extra={'readOnly': True, 'writeOnly': True, 'examples': ['postgresql://app:password@dbserver:5455/appdb', {'dsn': 'postgresql://app:password@dbserver:5455/appdb', 'schema_only': True}]})])

  • tablespace (Annotated[str | None, FieldInfo(annotation=NoneType, required=True, description='The name of the tablespace that will be associated with the database.'), AfterValidator(func=~pglift.models.interface.check_tablespace)])

class pglift.models.interface.Tablespace(*, name, location, size)
Parameters:

  • name (str)

  • location (str)

  • size (Annotated[int, <pglift.types.ByteSizeType object at 0x7ba0768f5be0>])

class pglift.models.interface.DefaultPrivilege(*, database, schema, object_type, role, privileges)

Default access privilege

Parameters:

  • database (str)

  • schema (str)

  • object_type (str)

  • role (str)

  • privileges (Annotated[list[str], AfterValidator(func=~pglift.models.interface._sort)])

class pglift.models.interface.Privilege(*, database, schema, object_type, role, privileges, object_name, column_privileges)

Access privilege

Parameters:
class pglift.postgresql.models.RewindSource(*, conninfo, password=None)

Configuration for pg_rewind.

Parameters:

  • conninfo (Annotated[str, Option(name='from', metavar='DSN'), FieldInfo(annotation=NoneType, required=True, description='DSN of source server to synchronize from.', exclude=True, json_schema_extra={'readOnly': True}), AfterValidator(func=functools.partial(<function check_conninfo at 0x7ba07633ae50>, exclude=['password']))])

  • password (Annotated[SecretStr | None, FieldInfo(annotation=NoneType, required=True, description='Password for the rewind user.', exclude=True, json_schema_extra={'readOnly': True})])

Results

class pglift.models.interface.ApplyResult(*, change_state=None, diff=<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)

Parameters:

  • change_state (Annotated[Literal['created', 'changed', 'dropped'] | None, FieldInfo(annotation=NoneType, required=True, description='Define the change applied (created, changed or dropped) to a manipulated object')])

  • diff (Any | None)