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.system.PostgreSQLInstance(name: str, version, settings: Settings)¶
A bare PostgreSQL instance.
- Raises:
InvalidVersion – if PostgreSQL executable directory (bindir) does not exist for specified version.
- 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.
- check() None ¶
Check if the instance exists and its configuration is valid.
- Raises:
InvalidVersion – if PG_VERSION content does not match declared version
InstanceNotFound – if PGDATA does not exist
InstanceNotFound – if configuration cannot be read
- config(managed_only: bool = False) Configuration ¶
Return parsed PostgreSQL configuration for this instance.
Refer to
pglift.conf.read()
for complete documentation.
- class pglift.models.system.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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- 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].
- model_fields: ClassVar[Dict[str, FieldInfo]] = {'datadir': FieldInfo(annotation=Path, required=True, description='PostgreSQL data directory.'), 'name': FieldInfo(annotation=str, required=True, description='Instance name.'), 'port': FieldInfo(annotation=int, required=True, description='TCP port the PostgreSQL instance is listening to.'), 'status': FieldInfo(annotation=str, required=True, description='Runtime status.'), 'version': FieldInfo(annotation=str, required=True, description='PostgreSQL version.')}¶
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.
This replaces Model.__fields__ from Pydantic V1.
- 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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- 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_fields: ClassVar[Dict[str, FieldInfo]] = {'auth': FieldInfo(annotation=Union[Auth, NoneType], required=False, default=None, exclude=True, json_schema_extra={'writeOnly': True}), 'creating': FieldInfo(annotation=bool, required=False, default=False, description='Internal field to indicate that the instance is being created.', exclude=True, validate_default=True, metadata=[Hidden(), Hidden(), AfterValidator(func=<function _set_creating_when_upgrading_from>)]), 'data_checksums': FieldInfo(annotation=Union[bool, NoneType], required=False, default=None, description='Enable or disable data checksums. If unspecified, fall back to site settings choice.'), 'databases': FieldInfo(annotation=list[Database], required=False, default=[], description='Databases defined in this instance (non-exhaustive list).', exclude=True, json_schema_extra={'writeOnly': True}, metadata=[Hidden()]), 'encoding': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Character encoding of the PostgreSQL instance.', json_schema_extra={'readOnly': True}), 'locale': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Default locale.', json_schema_extra={'readOnly': True}), 'name': FieldInfo(annotation=str, required=True, description='Instance name.', json_schema_extra={'readOnly': True}, metadata=[_PydanticGeneralMetadata(pattern='^[^/-]+$')]), 'pending_restart': FieldInfo(annotation=bool, required=False, default=False, description='Whether the instance needs a restart to account for settings changes.', json_schema_extra={'readOnly': True}, metadata=[Hidden(), Hidden()]), 'port': FieldInfo(annotation=Union[Annotated[int, AfterValidator], NoneType], required=False, default=None, description='TCP port the PostgreSQL instance will be listening to.', validate_default=True, metadata=[AfterValidator(func=<function _port_unset_is_available>)]), 'replication_slots': FieldInfo(annotation=list[Annotated[ReplicationSlot, BeforeValidator, WrapSerializer]], required=False, default=[], description='Replication slots in this instance (non-exhaustive list).', json_schema_extra={'examples': [{'name': 'myslot'}, 'someslot']}, metadata=[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'}), AfterValidator(func=<function _replication_slots_not_with_demoted_standby>)]), 'replrole_password': FieldInfo(annotation=Union[SecretStr, NoneType], required=False, default=None, description='Replication role password.', exclude=True, json_schema_extra={'readOnly': True}, metadata=[Option(name=None, metavar='password')]), 'restart_on_changes': FieldInfo(annotation=bool, required=False, default=False, description='Whether or not to automatically restart the instance to account for settings changes.', exclude=True, json_schema_extra={'writeOnly': True}, metadata=[Hidden()]), 'roles': FieldInfo(annotation=list[Role], required=False, default=[], description='Roles defined in this instance (non-exhaustive list).', exclude=True, json_schema_extra={'writeOnly': True}, metadata=[Hidden()]), 'settings': FieldInfo(annotation=MutableMapping[str, Any], required=False, default={}, 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'}]}, metadata=[AfterValidator(func=<function _no_port_in_settings>), Hidden()]), 'standby': FieldInfo(annotation=Union[Standby, NoneType], required=False, default=None, description='Standby information.'), 'state': FieldInfo(annotation=Literal['stopped', 'started', 'absent', 'restarted'], required=False, default='started', description='Runtime state.', metadata=[Choices(name=None, choices=['started', 'stopped'])]), 'surole_password': FieldInfo(annotation=Union[SecretStr, NoneType], required=False, default=None, description='Super-user role password.', exclude=True, json_schema_extra={'readOnly': True}, validate_default=True, metadata=[Option(name=None, metavar='password'), AfterValidator(func=<function _password_required_for_local_auth>)]), 'upgrading_from': FieldInfo(annotation=Union[PostgreSQLInstanceRef, NoneType], required=False, default=None, description='Internal field to keep a reference to the instance this manifest will be upgraded from.', exclude=True, metadata=[Hidden(), Hidden()]), 'version': FieldInfo(annotation=PostgreSQLVersion, 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, metadata=[BeforeValidator(func=<function _default_version>, json_schema_input_type=PydanticUndefined)])}¶
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.
This replaces Model.__fields__ from Pydantic V1.
- 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 0x7f93c51a4400>, '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, validity: ~datetime.datetime | None = None, memberships: ~typing.Annotated[list[~typing.Annotated[~pglift.models.interface.RoleMembership, ~pydantic.functional_validators.BeforeValidator(func=functools.partial(<function as_dict at 0x7f93c51602c0>, key='role'), json_schema_input_type=PydanticUndefined), ~pydantic.functional_serializers.WrapSerializer(func=functools.partial(<function serialize at 0x7f93c5163c40>, 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.'})] = [], in_roles: ~typing.Annotated[list[str], Hidden(), ~pydantic.functional_validators.AfterValidator(func=functools.partial(<function check_mutually_exclusive_with at 0x7f93c51a4400>, 'memberships', operations={'create', 'update'}))] = [], hba_records: ~typing.Annotated[list[~pglift.models.interface.HbaRecordForRole], Hidden()] = [], **extra_data: ~typing.Any)¶
PostgreSQL role
- model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- 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_fields: ClassVar[Dict[str, FieldInfo]] = {'connection_limit': FieldInfo(annotation=Union[int, NoneType], required=False, default=None, description='How many concurrent connections the role can make.'), 'createdb': FieldInfo(annotation=bool, required=False, default=False, description='Whether role can create new databases.'), 'createrole': FieldInfo(annotation=bool, required=False, default=False, description='Whether role can create new roles.'), 'drop_owned': FieldInfo(annotation=bool, required=False, default=False, description="Drop all PostgreSQL's objects owned by the role being dropped.", exclude=True, metadata=[Hidden(), AfterValidator(func=<function validate_state_is_absent>)]), 'encrypted_password': FieldInfo(annotation=Union[SecretStr, NoneType], required=False, default=None, description='Role password, already encrypted.', exclude=True, metadata=[AfterValidator(func=functools.partial(<function check_mutually_exclusive_with>, 'password'))]), 'has_password': FieldInfo(annotation=bool, required=False, default=False, description='True if the role has a password.', json_schema_extra={'readOnly': True}, validate_default=True, metadata=[Hidden(), Hidden(), AfterValidator(func=<function _set_has_password>)]), 'hba_records': FieldInfo(annotation=list[HbaRecordForRole], required=False, default=[], description='Entries in the pg_hba.conf file for this role.', metadata=[Hidden()]), 'in_roles': FieldInfo(annotation=list[str], required=False, default=[], description="DEPRECATED. Use 'memberships' instead.", metadata=[Hidden(), AfterValidator(func=functools.partial(<function check_mutually_exclusive_with>, 'memberships', operations={'create', 'update'}))]), 'inherit': FieldInfo(annotation=bool, required=False, default=True, description='Let the role inherit the privileges of the roles it is a member of.'), 'login': FieldInfo(annotation=bool, required=False, default=False, description='Allow the role to log in.'), 'memberships': FieldInfo(annotation=list[Annotated[RoleMembership, BeforeValidator, WrapSerializer]], required=False, default=[], description='Roles which this role should be a member of.', metadata=[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.'})]), 'name': FieldInfo(annotation=str, required=True, description='Role name.', json_schema_extra={'readOnly': True}), 'password': FieldInfo(annotation=Union[SecretStr, NoneType], required=False, default=None, description='Role password.', exclude=True), 'reassign_owned': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, 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=<function validate_state_is_absent>)]), 'replication': FieldInfo(annotation=bool, required=False, default=False, description='Whether the role is a replication role.'), 'state': FieldInfo(annotation=Literal['present', 'absent'], required=False, default='present', description='Whether the role be present or absent.', exclude=True, metadata=[Hidden()]), 'superuser': FieldInfo(annotation=bool, required=False, default=False, description='Whether the role is a superuser.'), 'validity': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None, description="Date and time after which the role's password is no longer valid.")}¶
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.
This replaces Model.__fields__ from Pydantic V1.
- 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: '})] = [], 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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- 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].
- model_fields: ClassVar[Dict[str, FieldInfo]] = {'clone': FieldInfo(annotation=Union[CloneOptions, NoneType], required=False, default=None, 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}]}), 'extensions': FieldInfo(annotation=list[Annotated[Extension, BeforeValidator, WrapSerializer]], required=False, default=[], description='Extensions in this database.', json_schema_extra={'examples': [{'name': 'unaccent', 'schema': 'ext', 'version': '1.0'}, 'hstore']}, metadata=[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.'})]), 'force_drop': FieldInfo(annotation=bool, required=False, default=False, description='Force the drop.', exclude=True, metadata=[Hidden(), AfterValidator(func=<function validate_state_is_absent>)]), 'name': FieldInfo(annotation=str, required=True, description='Database name.', json_schema_extra={'readOnly': True, 'examples': ['demo']}), 'owner': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='The role name of the user who will own the database.', json_schema_extra={'examples': ['postgres']}), 'publications': FieldInfo(annotation=list[Publication], required=False, default=[], description='Publications in this database.', json_schema_extra={'examples': [{'name': 'mypub'}]}, metadata=[Hidden()]), 'schemas': FieldInfo(annotation=list[Annotated[Schema, BeforeValidator, WrapSerializer]], required=False, default=[], description='Schemas in this database.', json_schema_extra={'examples': [{'name': 'sales'}, 'accounting']}, metadata=[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.'}), AfterValidator(func=<function _set_schemas_owner>)]), 'settings': FieldInfo(annotation=Union[MutableMapping[str, Union[str, bool, float, int, timedelta, NoneType]], NoneType], required=False, default=None, 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'}]}, metadata=[Hidden(), Spec(spec={'type': 'dict', 'required': False})]), 'state': FieldInfo(annotation=Literal['present', 'absent'], required=False, default='present', description='Database state.', exclude=True, json_schema_extra={'examples': ['present']}, metadata=[Hidden()]), 'subscriptions': FieldInfo(annotation=list[Subscription], required=False, default=[], description='Subscriptions in this database.', json_schema_extra={'examples': [{'name': 'mysub', 'publications': ['mypub'], 'enabled': False}]}, metadata=[Hidden()]), 'tablespace': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='The name of the tablespace that will be associated with the database.', metadata=[AfterValidator(func=<function check_tablespace>)])}¶
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.
This replaces Model.__fields__ from Pydantic V1.
- class pglift.models.interface.Tablespace(*, name: str, location: str, size: ~typing.Annotated[int, <pglift.types.ByteSizeType object at 0x7f93c631c5f0>])¶
- model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- 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].
- model_fields: ClassVar[Dict[str, FieldInfo]] = {'location': FieldInfo(annotation=str, required=True), 'name': FieldInfo(annotation=str, required=True), 'size': FieldInfo(annotation=int, required=True, metadata=[<pglift.types.ByteSizeType object>])}¶
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.
This replaces Model.__fields__ from Pydantic V1.
- 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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- 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].
- model_fields: ClassVar[Dict[str, FieldInfo]] = {'database': FieldInfo(annotation=str, required=True), 'object_type': FieldInfo(annotation=str, required=True), 'privileges': FieldInfo(annotation=list[str], required=True, metadata=[AfterValidator(func=<function _sort>)]), 'role': FieldInfo(annotation=str, required=True), 'schema_': FieldInfo(annotation=str, required=True, alias='schema', alias_priority=2)}¶
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.
This replaces Model.__fields__ from Pydantic V1.
- 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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- 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].
- model_fields: ClassVar[Dict[str, FieldInfo]] = {'column_privileges': FieldInfo(annotation=Mapping[str, list[str]], required=True, metadata=[AfterValidator(func=<function _sort_values>)]), 'database': FieldInfo(annotation=str, required=True), 'object_name': FieldInfo(annotation=str, required=True), 'object_type': FieldInfo(annotation=str, required=True), 'privileges': FieldInfo(annotation=list[str], required=True, metadata=[AfterValidator(func=<function _sort>)]), 'role': FieldInfo(annotation=str, required=True), 'schema_': FieldInfo(annotation=str, required=True, alias='schema', alias_priority=2)}¶
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.
This replaces Model.__fields__ from Pydantic V1.
Results¶
- class pglift.models.interface.ApplyResult(*, change_state: Literal['created', 'changed', 'dropped'] | None = None)¶
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)