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.InstanceListItem(name: str, version: str, port: int, datadir: Path, status: str)

class pglift.models.system.Database(name: str, owner: str, encoding: str, collation: str, ctype: str, acls: list[str], size, description: str | None, tablespace: Tablespace)

class pglift.models.system.Tablespace(name: str, location: str, size)

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

Default access privilege

class pglift.models.system.Privilege(database: str, schema: str, object_type: str, role: str, privileges: list[str], object_name: str, column_privileges: dict[str, list[str]])

Access privilege

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

Base class for manifest data classes.

classmethod parse_yaml(value: str | IO[str]) → Self

Parse from a YAML stream.

yaml(**kwargs: Any) → str

Return a YAML serialization of this manifest.

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]] = {}

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

service_manifest(stype: type[_S]) → _S

Return satellite service manifest 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_fields: ClassVar[dict[str, FieldInfo]] = {'auth': FieldInfo(annotation=Union[Auth, NoneType], required=False, exclude=True, json_schema_extra={'writeOnly': True}), 'data_checksums': FieldInfo(annotation=Union[bool, NoneType], required=False, 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)]), 'encoding': FieldInfo(annotation=Union[str, NoneType], required=False, description='Character encoding of the PostgreSQL instance.', json_schema_extra={'readOnly': True}), 'locale': FieldInfo(annotation=Union[str, NoneType], required=False, description='Default locale.', json_schema_extra={'readOnly': True}), 'name': FieldInfo(annotation=str, required=True, description='Instance name.', json_schema_extra={'readOnly': True}), '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), 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'.", metadata=[<pglift.types.PortValidatorType object>]), 'replrole_password': FieldInfo(annotation=Union[SecretStr, NoneType], required=False, 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)]), '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)]), '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)]), 'standby': FieldInfo(annotation=Union[Standby, NoneType], required=False, 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'])]), 'surole_password': FieldInfo(annotation=Union[SecretStr, NoneType], required=False, description='Super-user role password.', exclude=True, json_schema_extra={'readOnly': True}), 'version': FieldInfo(annotation=Union[PostgreSQLVersion, NoneType], required=False, 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.

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_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, 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)]), 'encrypted_password': FieldInfo(annotation=Union[SecretStr, NoneType], required=False, description='Role password, already encrypted.', exclude=True), '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), AnsibleConfig(hide=True, choices=None, spec=None)]), '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)]), '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, description='Role password.', exclude=True), 'reassign_owned': FieldInfo(annotation=Union[str, NoneType], required=False, 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)]), '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)]), 'superuser': FieldInfo(annotation=bool, required=False, default=False, description='Whether the role is a superuser.'), 'validity': FieldInfo(annotation=Union[datetime, NoneType], required=False, 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.

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)

change_state: Literal['created', 'changed', 'dropped'] | None