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.

classmethod system_lookup(value: BaseInstance | tuple[str, str | None, Settings]) → Self¶

Build a (real) instance by system lookup.

Parameters:: value – either a BaseInstance object or a (name, version) tuple.
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:: 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.

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.

class pglift.models.system.Instance(name: str, version, settings: Settings, services: list[Any])¶

A PostgreSQL instance with satellite services.

classmethod system_lookup(value: BaseInstance | tuple[str, str | None, Settings]) → Self¶

Build a (real) instance by system lookup.

Parameters:: value – either a BaseInstance object or a (name, version) tuple.
Raises:: InstanceNotFound – if the instance could not be found by system lookup.

service(stype: type[S]) → S¶

Return bound satellite service object matching requested type.

Raises:: ValueError – if not found.

class pglift.models.system.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_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].

This replaces Model.__fields__ from Pydantic V1.

class pglift.models.interface.Instance(*, name: str, version: PostgreSQLVersion | None = None, port: int = 5432, settings: MutableMapping[str, Any] = {}, surole_password: SecretStr | None = None, replrole_password: SecretStr | None = None, data_checksums: bool | None = None, locale: str | None = None, encoding: str | None = None, auth: Auth | None = None, standby: Standby | None = None, state: Literal['stopped', 'started', 'absent', 'restarted'] = 'started', databases: list[Database] = [], roles: list[Role] = [], pending_restart: bool = False, restart_on_changes: bool = False, **extra_data: Any)¶

PostgreSQL instance

>>> Instance(name='without_dash')  
Instance(name='without_dash', ...)
>>> Instance(name='with-dash')
Traceback (most recent call last):
    ...
pydantic_core._pydantic_core.ValidationError: 1 validation error for Instance
name
  Value error, must not contain dashes [type=value_error, input_value='with-dash', input_type=str]
  ...
>>> Instance(name='with/slash')
Traceback (most recent call last):
    ...
pydantic_core._pydantic_core.ValidationError: 1 validation error for Instance
name
  Value error, must not contain slashes [type=value_error, input_value='with/slash', input_type=str]
  ...

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}), '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=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False)]), '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=[AfterValidator(func=functools.partial(<function check_excludes>, ('/', 'slashes'))), AfterValidator(func=functools.partial(<function check_excludes>, ('-', 'dashes')))]), '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=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False), AnsibleConfig(hide=True, choices=None, spec=None)]), 'port': FieldInfo(annotation=int, required=False, default=5432, description="TCP port the PostgreSQL instance will be listening to. If unspecified, default to 5432 unless a 'port' setting is found in 'settings'.", validate_default=True, metadata=[AfterValidator(func=<function check_port_available>)]), 'replrole_password': FieldInfo(annotation=Union[SecretStr, NoneType], required=False, default=None, description='Replication role password.', exclude=True, json_schema_extra={'readOnly': True}), '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=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False)]), '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=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False)]), '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=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False)]), '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=[CLIConfig(name=None, hide=False, metavar=None, choices=['started', 'stopped'], as_option=False)]), 'surole_password': FieldInfo(annotation=Union[SecretStr, NoneType], required=False, default=None, description='Super-user role password.', exclude=True, json_schema_extra={'readOnly': True}), 'version': FieldInfo(annotation=Union[PostgreSQLVersion, NoneType], required=False, default=None, description='PostgreSQL version.', json_schema_extra={'readOnly': True})}¶

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

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(stype: type[_S]) → _S¶

Return satellite Service attached to this instance.

Raises:: ValueError – if not found.

class pglift.models.interface.Role(*, name: str, state: Literal['present', 'absent'] = 'present', password: SecretStr | None = None, encrypted_password: SecretStr | None = None, drop_owned: bool = False, reassign_owned: str | None = None, has_password: bool = 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 | None = None, in_roles: list[str] = [], **extra_data: 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=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False), 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=<function check_one_password_only>)]), '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=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False), AnsibleConfig(hide=True, choices=None, spec=None), AfterValidator(func=<function _set_has_password>)]), 'in_roles': FieldInfo(annotation=list[str], required=False, default=[], description='List of roles to which the new role will be added as a new member.', metadata=[CLIConfig(name='in_role', hide=False, metavar=None, choices=None, as_option=False)]), '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.'), '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, metadata=[AfterValidator(func=<function check_one_password_only>)]), '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), CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False), 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=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False)]), '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].

This replaces Model.__fields__ from Pydantic V1.

class pglift.models.interface.Database(*, name: str, force_drop: bool = False, state: Literal['present', 'absent'] = 'present', owner: str | None = None, settings: MutableMapping[str, str | bool | float | int | timedelta | None] | None = None, schemas: list[Schema] = [], extensions: list[Extension] = [], publications: list[Publication] = [], subscriptions: list[Subscription] = [], clone: CloneOptions | None = None, tablespace: str | None = 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]], required=False, default=[], description='List of extensions to create in the database.', json_schema_extra={'examples': [{'name': 'unaccent', 'schema': 'ext', 'version': '1.0'}, 'hstore']}, metadata=[CLIConfig(name='extension', hide=False, metavar=None, choices=None, as_option=False)]), 'force_drop': FieldInfo(annotation=bool, required=False, default=False, description='Force the drop.', exclude=True, metadata=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False), 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='List of publications to in the database.', json_schema_extra={'examples': [{'name': 'mypub'}]}, metadata=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False)]), 'schemas': FieldInfo(annotation=list[Annotated[Schema, BeforeValidator]], required=False, default=[], description='List of schemas to create in the database.', json_schema_extra={'examples': [{'name': 'sales'}, 'accounting']}, metadata=[CLIConfig(name='schema', hide=False, metavar=None, choices=None, as_option=False)]), '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=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False), AnsibleConfig(hide=False, choices=None, 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=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False)]), 'subscriptions': FieldInfo(annotation=list[Subscription], required=False, default=[], description='List of subscriptions to in the database.', json_schema_extra={'examples': [{'name': 'mysub', 'publications': ['mypub'], 'enabled': False}]}, metadata=[CLIConfig(name=None, hide=True, metavar=None, choices=None, as_option=False)]), '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].

This replaces Model.__fields__ from Pydantic V1.

class pglift.models.interface.Tablespace(*, name: str, location: str, size: int)¶

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

This replaces Model.__fields__ from Pydantic V1.

class pglift.models.interface.DefaultPrivilege(*, database: str, schema: str, object_type: str, role: str, privileges: list[str])¶

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

This replaces Model.__fields__ from Pydantic V1.

class pglift.models.interface.Privilege(*, database: str, schema: str, object_type: str, role: str, privileges: list[str], object_name: str, column_privileges: Mapping[str, list[str]])¶

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

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)