Programmatic API reference

This page describes parts of inmanta code base that provide a stable API that could be used from modules or extensions.

Warning

Only those parts explicitly mentioned here are part of the API. They provide a stable interface. Other parts of the containing modules provide no such guarantees.

Constants

class inmanta.const.LogLevel(*values)

Bases: str, Enum

Log levels used for various parts of the inmanta orchestrator.

CRITICAL = 'CRITICAL'

DEBUG = 'DEBUG'

ERROR = 'ERROR'

INFO = 'INFO'

TRACE = 'TRACE'

WARNING = 'WARNING'

property to_int: int

class inmanta.const.ResourceAction(*values)

Bases: str, Enum

Enumeration of all resource actions.

deploy = 'deploy'

dryrun = 'dryrun'

getfact = 'getfact'

other = 'other'

pull = 'pull'

push = 'push'

store = 'store'

Compiler exceptions

class inmanta.ast.CompilerException(msg: str)

Bases: Exception, Exportable

Base class for exceptions generated by the compiler

class inmanta.parser.ParserException(location: Location, value: object, msg: str | None = None)

Bases: CompilerException

Exception occurring during the parsing of the code

class inmanta.ast.RuntimeException(stmt: Locatable | None, msg: str)

Bases: CompilerException

Baseclass for exceptions raised by the compiler after parsing is complete.

class inmanta.ast.ExternalException(stmt: Locatable | None, msg: str, cause: Exception)

Bases: RuntimeException

When a plugin call produces an exception that is not a RuntimeException, it is wrapped in an ExternalException to make it conform to the expected interface

class inmanta.ast.ExplicitPluginException(stmt: Locatable | None, msg: str, cause: PluginException)

Bases: ExternalException

Base exception for wrapping an explicit inmanta.plugins.PluginException raised from a plugin call.

Plugins

class inmanta.plugins.Context(resolver: Resolver, queue: QueueScheduler, owner: FunctionCall, plugin: Plugin, result: ResultVariable)

An instance of this class is used to pass context to the plugin

get_client() → Client

get_compiler() → Compiler

get_data_dir() → str

Get the path to the data dir (and create if it does not exist yet

get_environment_id() → str

get_queue_scheduler() → QueueScheduler

get_resolver() → Resolver

get_sync_client() → SyncClient

get_type(name: LocatableString) → Type

Get a type from the configuration model.

run_sync(function: Callable[[], Awaitable[T]], timeout: int = 5) → T

Execute the async function and return its result. This method uses this thread’s current (not running) event loop if there is one, otherwise it creates a new one. The main use for this function is to use the inmanta internal rpc to communicate with the server.

Parameters:

• function – The async function to execute. This function should return a yieldable object.

• timeout – A timeout for the async function.

Returns:

The result of the async call.

Raises:

ConnectionRefusedError – When the function timeouts this exception is raised.

inmanta.plugins.plugin(function: str | None = None, commands: list[str] | None = None, emits_statements: bool = False, allow_unknown: bool = False) → Callable[[T_FUNC], T_FUNC]

inmanta.plugins.plugin(function: T_FUNC, commands: list[str] | None = None, emits_statements: bool = False, allow_unknown: bool = False) → T_FUNC

Python decorator to register functions with inmanta as plugin

Parameters:

• function – The function to register with inmanta. This is the first argument when it is used as decorator.

• commands – A list of command paths that need to be available. Inmanta raises an exception when the command is not available.

• emits_statements – Set to true if this plugin emits new statements that the compiler should execute. This is only required for complex plugins such as integrating a template engine.

• allow_unknown – Set to true if this plugin accepts Unknown values as valid input.

class inmanta.plugins.PluginException(message: str)

Base class for custom exceptions raised from a plugin.

class inmanta.plugins.PluginMeta(name: str, bases: tuple[type, ...], dct: dict[str, object])

Bases: type

A metaclass that keeps track of concrete plugin subclasses. This class is responsible for all plugin registration.

classmethod add_function(plugin_class: type[Plugin]) → None

Add a function plugin class

classmethod clear(inmanta_module: str | None = None) → None

Clears registered plugin functions.

Parameters:

inmanta_module – Clear plugin functions for a specific inmanta module. If omitted, clears all registered plugin functions.

classmethod get_functions() → dict[str, Type[Plugin]]

Get all functions that are registered

class inmanta.ast.UnsetException(msg: str, instance: Instance | None = None, attribute: Attribute | None = None)

This exception is thrown when an attribute is accessed that was not yet available (i.e. it has not been frozen yet).

inmanta.plugins.allow_reference_values(instance: T) → T

For the given DSL instance, or list or dict nested inside an instance, allow accessing undeclared reference values (attributes, list elements or dict values respectively). Reference values are otherwise rejected on access because not all plugins can be assumed to be compatible with them, and may reasonably expect values of the DSL attributes’ declared type.

Does not allow nested access. Each object for which reference elements are expected should be wrapped separately, e.g. allow_reference_values(my_instance.my_relation).maybe_reference rather than allow_reference_values(my_instance).my_relation.maybe_reference.

This function is not required for plugin arguments or dataclasses. In those cases, reference support can be declared directly via type annotations (e.g. int | Reference[int]), in which case no special access function is required. However, when called on such values, the function simply returns the argument unchanged.

Resources

inmanta.resources.resource(name: str, id_attribute: str, agent: str)

A decorator that registers a new resource. The decorator must be applied to classes that inherit from Resource

Parameters:

• name – The name of the entity in the configuration model it creates a resource from. For example std::testing::NullResource

• id_attribute – The attribute of this resource that uniquely identifies a resource on a logical agent. This attribute can be mapped.

• agent – This string indicates how the agent of this resource is determined. This string points to an attribute, but it can navigate relations (this value cannot be mapped). For example, the agent argument could be host.name.

class inmanta.resources.Resource(_id: Id)

Plugins should inherit resource from this class so a resource from a model can be serialized and deserialized.

Such as class is registered when the resource() decorator is used. Each class needs to indicate the fields the resource will have with a class field named “fields”. A metaclass merges all fields lists from the class itself and all superclasses. If a field it not available directly in the model object the serializer will look for static methods in the class with the name “get_$fieldname”.

clone(**kwargs: Any) → R

Create a clone of this resource. The given kwargs can be used to override attributes.

Returns:

The cloned resource

class inmanta.resources.PurgeableResource(_id: Id)

See std::PurgeableResource for more information.

class inmanta.resources.ManagedResource(_id: Id)

See std::ManagedResource for more information.

class inmanta.resources.IgnoreResourceException

Throw this exception when a resource should not be included by the exported. Typically resources use this to indicate that they are not managed by the orchestrator.

class inmanta.resources.Id(entity_type: str, agent_name: str, attribute: str, attribute_value: str, version: int = 0)

A unique id that identifies a resource that is managed by an agent

classmethod parse_id(resource_id: ResourceVersionIdStr | ResourceIdStr, version: int | None = None) → Id

Parse the resource id and return the type, the hostname and the resource identifier.

Parameters:

version – If provided, the version field of the returned Id will be set to this version.

resource_str() → ResourceIdStr

String representation for this resource id with the following format:

<type>[<agent>,<attribute>=<value>]

• type: The resource type, as defined in the configuration model.

  For example std::testing::NullResource.

• agent: The agent responsible for this resource.

• attribute: The key attribute that uniquely identifies this resource on the agent

• value: The corresponding value for this key attribute.

Returns:

Returns a inmanta.data.model.ResourceIdStr

class inmanta.execute.util.Unknown(source: object)

An instance of this class is used to indicate that this value can not be determined yet.

Parameters:

source – The source object that can determine the value

Handlers

inmanta.agent.handler.cache(func: None = None, ignore: list[str] = [], timeout: int | None = None, for_version: bool | None = None, cache_none: bool = True, cacheNone: bool | None = None, call_on_delete: Callable[[Any], None] | None = None, evict_after_creation: float = 0.0, evict_after_last_access: float = 0.0) → Callable[[T_FUNC], T_FUNC]

inmanta.agent.handler.cache(func: T_FUNC, ignore: list[str] = [], timeout: int | None = None, for_version: bool | None = None, cache_none: bool = True, cacheNone: bool | None = None, call_on_delete: Callable[[Any], None] | None = None, evict_after_creation: float = 0.0, evict_after_last_access: float = 0.0) → T_FUNC

decorator for methods in resource handlers to provide caching

this decorator works similar to memoization: when the decorate method is called, its return value is cached, for subsequent calls, the cached value is used instead of the actual value

The name of the method + the arguments of the method form the cache key

If an argument named resource is present, it is assumed to be a resource and its ID is used, without the version information

Parameters:

• ignore – a list of argument names that should not be part of the cache key

• cache_none – allow the caching of None values

• call_on_delete – A callback function that is called when the value is removed from the cache, with the value as argument.

• evict_after_creation – This cache item will be considered stale this number of seconds after entering the cache.

• evict_after_last_access – This cache item will be considered stale this number of seconds after it was last accessed.

inmanta.agent.handler.provider(resource_type: str, name: str | None = None) → None

A decorator that registers a new handler.

Parameters:

• resource_type – The type of the resource this handler is responsible for. For example, std::testing::NullResource

• name – A name to reference this provider.

class inmanta.agent.handler.SkipResource

Bases: Exception

A handler should raise this exception when a resource should be skipped. The resource will be marked as skipped instead of failed. We will try to deploy again later .

class inmanta.agent.handler.SkipResourceForDependencies

Bases: SkipResource

A handler should raise this exception when a resource should be skipped as a result of unsuccessful dependencies. The resource will be marked as skipped instead of failed. We will try to deploy again when its dependencies are successfully deployed for their latest intent.

class inmanta.agent.handler.ResourcePurged

If the read_resource() method raises this exception, the agent will mark the current state of the resource as purged.

class inmanta.agent.handler.HandlerContext(resource: Resource, dry_run: bool = False, action_id: UUID | None = None, logger: Logger | None = None)

Context passed to handler methods for state related “things”

add_change(name: str, desired: object, current: object = None) → None

Report a change of a field. This field is added to the set of updated fields

Parameters:

• name – The name of the field that was updated

• desired – The desired value to which the field was updated (or should be updated)

• current – The value of the field before it was updated

add_changes(**kwargs: SimpleTypes) → None

Report a list of changes at once as kwargs

Parameters:

• key – The name of the field that was updated. This field is also added to the set of updated fields

• value – The desired value of the field.

To report the previous value of the field, use the add_change method

fields_updated(fields: str) → None

Report that fields have been updated

is_dry_run() → bool

Is this a dryrun?

set_fact(fact_id: str, value: str, expires: bool = True) → None

Send a fact to the Inmanta server.

Parameters:

• fact_id – The name of the fact.

• value – The actual value of the fact.

• expires – Whether this fact expires or not.

set_resource_state(new_state: HandlerResourceState) → None

Set the state of the resource

set_status(status: ResourceState) → None

Set the status of the handler operation and translate it to HandlerResourceState

update_changes(changes: dict[str, AttributeStateChange]) → None

update_changes(changes: dict[str, dict[str, SimpleTypes | None]]) → None

update_changes(changes: dict[str, tuple[SimpleTypes, SimpleTypes]]) → None

Update the changes list with changes

Parameters:

changes – This should be a dict with a value a dict containing “current” and “desired” keys

class inmanta.agent.handler.ResourceHandler(agent: AgentInstance, io: object = None)

A class that handles resources.

_abc_impl = <_abc._abc_data object>

_diff(current: TResource, desired: TResource) → dict[str, dict[str, Any]]

Calculate the diff between the current and desired resource state.

Parameters:

• current – The current state of the resource

• desired – The desired state of the resource

Returns:

A dict with key the name of the field and value another dict with “current” and “desired” as keys for fields that require changes.

check_facts(ctx: HandlerContext, resource: TResource) → dict[str, str]

This method is called by the agent to query for facts. It runs pre() and post(). This method calls facts() to do the actual querying.

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to query facts for.

Returns:

A dict with fact names as keys and facts values.

check_resource(ctx: HandlerContext, resource: TResource) → TResource

Check the current state of a resource

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to check the current state of.

Returns:

A resource to represents the current state. Use the clone() to create clone of the given resource that can be modified.

do_changes(ctx: HandlerContext, resource: TResource, changes: Mapping[str, Mapping[str, object]]) → None

Do the changes required to bring the resource on this system in the state of the given resource.

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to check the current state of.

• changes – The changes that need to occur as reported by list_changes()

execute(ctx: HandlerContext, resource: TResource, dry_run: bool = False) → None

Enforce a resource’s intent and inform the handler context of any relevant changes (e.g. set deployed status, report attribute changes). Called only when all of its dependencies have successfully deployed.

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to deploy.

• dry_run – If set to true, the intent is not enforced, only the set of changes it would bring is computed.

list_changes(ctx: HandlerContext, resource: TResource) → dict[str, dict[str, Any]]

Returns the changes required to bring the resource on this system in the state described in the resource entry. This method calls check_resource()

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to check the current state of.

Returns:

A dict with key the name of the field and value another dict with “current” and “desired” as keys for fields that require changes.

class inmanta.agent.handler.CRUDHandler(agent: AgentInstance, io: object = None)

This handler base class requires CRUD methods to be implemented: create, read, update and delete. Such a handler only works on purgeable resources.

calculate_diff(ctx: HandlerContext, current: TPurgeableResource, desired: TPurgeableResource) → dict[str, dict[str, Any]]

Calculate the diff between the current and desired resource state.

Parameters:

• ctx – Context can be used to get values discovered in the read method. For example, the id used in API calls. This context should also be used to let the handler know what changes were made to the resource.

• current – The current state of the resource

• desired – The desired state of the resource

Returns:

A dict with key the name of the field and value another dict with “current” and “desired” as keys for fields that require changes.

can_reload() → bool

Can this handler reload?

Returns:

Return true if this handler needs to reload on requires changes.

check_facts(ctx: HandlerContext, resource: TResource) → dict[str, str]

This method is called by the agent to query for facts. It runs pre() and post(). This method calls facts() to do the actual querying.

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to query facts for.

Returns:

A dict with fact names as keys and facts values.

check_resource(ctx: HandlerContext, resource: TResource) → TResource

Check the current state of a resource

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to check the current state of.

Returns:

A resource to represents the current state. Use the clone() to create clone of the given resource that can be modified.

close() → None

Override this method to implement custom logic called by the agent on handler deactivation. i.e. when the instantiated handler will no longer be used by the agent.

create_resource(ctx: HandlerContext, resource: TPurgeableResource) → None

This method is called by the handler when the resource should be created.

Parameters:

• context – Context can be used to get values discovered in the read method. For example, the id used in API calls. This context should also be used to let the handler know what changes were made to the resource.

• resource – The desired resource state.

delete_resource(ctx: HandlerContext, resource: TPurgeableResource) → None

This method is called by the handler when the resource should be deleted.

Parameters:

• ctx – Context can be used to get values discovered in the read method. For example, the id used in API calls. This context should also be used to let the handler know what changes were made to the resource.

• resource – The desired resource state.

deploy(ctx: HandlerContext, resource: TResource, requires: Mapping[ResourceIdStr, ResourceState]) → None

Main entrypoint of the handler that will be called by the agent to deploy a resource on the server. The agent calls this method for a given resource as soon as all its dependencies (requires relation) are ready. It is always called, even when one of the dependencies failed to deploy.

Takes appropriate action based on the state of its dependencies. Calls execute iff the handler should actually execute, i.e. enforce the intent represented by the resource. A handler may choose not to proceed to this execution stage, e.g. when one of the resource’s dependencies failed.

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to deploy

• requires – A dictionary mapping the resource id of each dependency of the given resource to its resource state.

do_changes(ctx: HandlerContext, resource: TResource, changes: Mapping[str, Mapping[str, object]]) → None

Do the changes required to bring the resource on this system in the state of the given resource.

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to check the current state of.

• changes – The changes that need to occur as reported by list_changes()

do_reload(ctx: HandlerContext, resource: TResource) → None

Perform a reload of this resource.

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to reload.

execute(ctx: HandlerContext, resource: TPurgeableResource, dry_run: bool = False) → None

Enforce a resource’s intent and inform the handler context of any relevant changes (e.g. set deployed status, report attribute changes). Called only when all of its dependencies have successfully deployed.

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to deploy.

• dry_run – If set to true, the intent is not enforced, only the set of changes it would bring is computed.

facts(ctx: HandlerContext, resource: TResource) → dict[str, str]

Override this method to implement fact querying. A queried fact can be reported back in two different ways: either via the return value of this method or by adding the fact to the HandlerContext via the set_fact() method. pre() and post() are called before and after this method.

Parameters:

• ctx – Context object to report changes, logs and facts to the agent and server.

• resource – The resource to query facts for.

Returns:

A dict with fact names as keys and facts values.

get_client() → SessionClient

Get the client instance that identifies itself with the agent session.

Returns:

A client that is associated with the session of the agent that executes this handler.

get_file(hash_id: str) → bytes | None

Retrieve a file from the fileserver identified with the given id.

Parameters:

hash_id – The id of the content/file to retrieve from the server.

Returns:

The content in the form of a bytestring or none is the content does not exist.

list_changes(ctx: HandlerContext, resource: TResource) → dict[str, dict[str, Any]]

Returns the changes required to bring the resource on this system in the state described in the resource entry. This method calls check_resource()

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource to check the current state of.

Returns:

A dict with key the name of the field and value another dict with “current” and “desired” as keys for fields that require changes.

post(ctx: HandlerContext, resource: TResource) → None

Method executed after a handler operation. Override this method to run after an operation.

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource being handled.

pre(ctx: HandlerContext, resource: TResource) → None

Method executed before a handler operation (Facts, dryrun, real deployment, …) is executed. Override this method to run before an operation.

Parameters:

• ctx – Context object to report changes and logs to the agent and server.

• resource – The resource being handled.

read_resource(ctx: HandlerContext, resource: TPurgeableResource) → None

This method reads the current state of the resource. It provides a copy of the resource that should be deployed, the method implementation should modify the attributes of this resource to the current state.

Parameters:

• ctx – Context can be used to pass value discovered in the read method to the CUD methods. For example, the id used in API calls

• resource – A clone of the desired resource state. The read method need to set values on this object.

Raises:

• SkipResource – Raise this exception when the handler should skip this resource

• SkipResourceForDependencies – Raise this exception when the handler should skip this resource and retry only when its dependencies succeed.

• ResourcePurged – Raise this exception when the resource does not exist yet.

run_sync(func: Callable[[], Awaitable[T]]) → T

Run the given async function on the ioloop of the agent. It will block the current thread until the future resolves.

Parameters:

func – A function that returns a yieldable future.

Returns:

The result of the async function.

set_cache(cache: AgentCache) → None

The agent calls this method when it has deemed this handler suitable for a given resource. This cache will be used for methods decorated with @cache.

Parameters:

cache – The AgentCache to use.

stat_file(hash_id: str) → bool

Check if a file exists on the server.

Parameters:

hash_id – The id of the file on the server. The convention is the use the sha1sum of the content as id.

Returns:

True if the file is available on the server.

update_resource(ctx: HandlerContext, changes: dict[str, dict[str, Any]], resource: TPurgeableResource) → None

This method is called by the handler when the resource should be updated.

Parameters:

• ctx – Context can be used to get values discovered in the read method. For example, the id used in API calls. This context should also be used to let the handler know what changes were made to the resource.

• changes – A map of resource attributes that should be changed. Each value is a tuple with the current and the desired value.

• resource – The desired resource state.

upload_file(hash_id: str, content: bytes) → None

Upload a file to the server

Parameters:

• hash_id – The id to identify the content. The convention is to use the sha1sum of the content to identify it.

• content – A byte string with the content

References

class inmanta.references.Reference

Instances of this class can create references to a value and resolve them.

@inmanta.references.reference

This decorator registers a reference under a specific name

Export

@inmanta.export.dependency_manager(function: Callable[[dict[str, Entity], dict[Id, Resource]], None]) → None

Register a function that manages dependencies in the configuration model that will be deployed.

Attributes

class inmanta.ast.attribute.Attribute(entity: Entity, value_type: Type, name: str, location: Location, multi: bool = False, nullable: bool = False)

The attribute base class for entity attributes.

Parameters:

entity – The entity this attribute belongs to

get_type() → Type

Get the declared type of this attribute.

property type: Type

Get the declared type of this attribute.

validate(value: object) → None

Validate a value that is going to be assigned to this attribute. Raises a inmanta.ast.RuntimeException if validation fails.

class inmanta.ast.attribute.RelationAttribute(entity: Entity, value_type: Type, name: str, location: Location)

Bases: Attribute

An attribute that is a relation

Modules

class inmanta.module.InstallMode(*values)

Bases: str, Enum

The module install mode determines what version of a module should be selected when a module is downloaded.

master = 'master'

For V1 modules: Use the module’s master branch. For V2 modules: Equivalent to InstallMode.prerelease

prerelease = 'prerelease'

Similar to InstallMode.release but prerelease versions are allowed as well.

release = 'release'

Only use a released version that is compatible with the current compiler and any version constraints defined in the requires lists for the project or any other modules (see ProjectMetadata, ModuleV1Metadata and ModuleV2Metadata).

A module is considered released in the following situations:

• For V1 modules: There is a tag on a commit. This tag is a valid, pep440 compliant version identifier and it’s not a

  prelease version.

• For V2 modules: The python package was published on a Python package repository, the version identifier is pep440

  compliant and is not a prerelease version.

inmanta.module.INSTALL_OPTS: list[str] = ['release', 'prerelease', 'master']

List of possible module install modes, kept for backwards compatibility. New code should use InstallMode instead.

class inmanta.module.InvalidModuleException(msg: str)

This exception is raised if a module is invalid.

class inmanta.module.InvalidMetadata(msg: str, validation_error: ValidationError | None = None)

This exception is raised if the metadata file of a project or module is invalid.

class inmanta.module.ModuleLike(path: str)

Bases: ABC, Generic[TMetadata]

Commons superclass for projects and modules, which are both versioned by git

Variables:

name – The name for this module like instance, in the context of the Inmanta DSL.

abstractmethod classmethod from_path(path: str) → ModuleLike | None

Get a concrete module like instance from a path. Returns None when no project or module is present at the given path.

property metadata: TMetadata

class inmanta.module.Module(project: Project | None, path: str)

Bases: ModuleLike[TModuleMetadata], ABC

This class models an inmanta configuration module

abstractmethod classmethod from_path(path: str) → Module | None

Get a concrete module like instance from a path. Returns None when no project or module is present at the given path.

get_plugin_files() → Iterator[tuple[Path, ModuleName]]

Returns a tuple (absolute_path, fq_mod_name) of all python files in this module.

unload() → None

Unloads this module instance from the project, the registered plugins and the loaded Python modules.

inmanta.module.ModuleName = inmanta.module.ModuleName

NewType creates simple unique types with almost zero runtime overhead.

NewType(name, tp) is considered a subtype of tp by static type checkers. At runtime, NewType(name, tp) returns a dummy callable that simply returns its argument.

Usage:

UserId = NewType('UserId', int)

def name_by_id(user_id: UserId) -> str:
    ...

UserId('user')          # Fails type check

name_by_id(42)          # Fails type check
name_by_id(UserId(42))  # OK

num = UserId(5) + 1     # type: int

class inmanta.module.ModuleV1(project: Project | None, path: str)

Bases: Module[ModuleV1Metadata], ModuleLikeWithYmlMetadataFile

classmethod from_path(path: str) → TModule | None

Get a concrete module like instance from a path. Returns None when no project or module is present at the given path.

class inmanta.module.ModuleV2(project: Project | None, path: str, is_editable_install: bool = False, installed_version: Version | None = None)

Bases: Module[ModuleV2Metadata]

classmethod from_path(path: str) → TModule | None

Get a concrete module like instance from a path. Returns None when no project or module is present at the given path.

is_editable() → bool

Returns True iff this module has been installed in editable mode.

class inmanta.module.ModuleSource

Bases: Generic[TModule]

get_installed_module(project: Project | None, module_name: str) → TModule | None

Returns a module object for a module if it is installed.

Parameters:

• project – The project associated with the module.

• module_name – The name of the module.

class inmanta.module.ModuleV2Source(urls: Sequence[str] = [])

Bases: ModuleSource[ModuleV2]

inmanta.module.Path = inmanta.module.Path

NewType creates simple unique types with almost zero runtime overhead.

NewType(name, tp) is considered a subtype of tp by static type checkers. At runtime, NewType(name, tp) returns a dummy callable that simply returns its argument.

Usage:

UserId = NewType('UserId', int)

def name_by_id(user_id: UserId) -> str:
    ...

UserId('user')          # Fails type check

name_by_id(42)          # Fails type check
name_by_id(UserId(42))  # OK

num = UserId(5) + 1     # type: int

class inmanta.loader.PluginModuleFinder(modulepaths: list[str])

Bases: MetaPathFinder

Custom module finder which handles V1 Inmanta modules. V2 modules are handled using the standard Python finder. This finder is stored as the last entry in meta_path, as such that the default Python Finders detect V2 modules first.

classmethod reset() → None

Remove the PluginModuleFinder from sys.meta_path.

inmanta.loader.unload_inmanta_plugins(inmanta_module: str | None = None) → None

Unloads Python modules associated with inmanta modules (inmanta_plugins submodules).

Parameters:

inmanta_module – Unload the Python modules for a specific inmanta module. If omitted, unloads the Python modules for all inmanta modules.

Project

class inmanta.module.Project(path: str, autostd: bool = True, main_file: str = 'main.cf', venv_path: str | VirtualEnv | None = None, attach_cf_cache: bool = True, strict_deps_check: bool | None = None)

Bases: ModuleLike[ProjectMetadata], ModuleLikeWithYmlMetadataFile

An inmanta project

Variables:

• modules – The collection of loaded modules for this project.

• module_source – The v2 module source for this project.

classmethod get(main_file: str = 'main.cf') → Project

Get the instance of the project

install_modules(*, bypass_module_cache: bool = False, update: bool = False) → None

Installs all modules.

Parameters:

• bypass_module_cache – Fetch the module data from disk even if a cache entry exists.

• update – Update all modules and their Python dependencies (recursive) to their latest versions.

load(install: bool = False) → None

Load this project’s AST and plugins.

Parameters:

install – Whether to install the project’s modules before attempting to load it.

classmethod set(project: Project, *, clean: bool = True) → None

Set the instance of the project.

Parameters:

clean – Clean up all side effects of any previously loaded projects. Clears the registered plugins and loaded Python plugins packages.

class inmanta.module.ProjectNotFoundException(msg: str)

Bases: CompilerException

This exception is raised when inmanta is unable to find a valid project

Python Environment

inmanta.env.mock_process_env(*, python_path: str | None = None, env_path: str | None = None) → None

Overrides the process environment information. This forcefully sets the environment that is recognized as the outer Python environment. This function should only be called when a Python environment has been set up dynamically and this environment should be treated as if this process was spawned from it, and even then with great care. :param python_path: The path to the python binary. Only one of python_path and env_path should be set. :param env_path: The path to the python environment directory. Only one of python_path and env_path should be set.

When using this method in a fixture to set and reset virtualenv, it is preferable to use store_venv()

class inmanta.env.VirtualEnv(env_path: str)

Creates and uses a virtual environment for this process. This virtualenv inherits from the previously active one.

init_env() → None

Initialize the virtual environment.

use_virtual_env() → None

Activate the virtual environment.

Variables

class inmanta.ast.variables.Reference(name: LocatableString)

This class represents a reference to a value

Variables:

name – The name of the Reference as a string.

name

Typing

The inmanta.ast.type module contains a representation of inmanta types, as well as validation logic for those types.

class inmanta.ast.type.Type

This class is the abstract base class for all types in the Inmanta DSL that represent basic data. These are types that are not relations. Instances of subclasses represent a type in the Inmanta language.

get_base_type() → Type

Returns the base type for this type, i.e. the plain type without modifiers such as expressed by [] and ? in the DSL and Reference in the plugin domain.

type_string() → str | None

Returns the type string as expressed in the Inmanta DSL, if this type can be expressed in the DSL. Otherwise returns None.

validate(value: object | None) → bool

Validate the given value to check if it satisfies the constraints associated with this type. Returns true iff validation succeeds, otherwise raises a inmanta.ast.RuntimeException.

In advanced cases where this class has a custom to_python(), translation-specific validation should be deferred to that stage. Translation-specific means that the value is definitely of this DSL type, but it can not be converted to the Python domain.

with_base_type(base_type: Type) → Type

Returns the type formed by replacing this type’s base type with the supplied type.

class inmanta.ast.type.NullableType(element_type: Type)

Bases: Type

Represents a nullable type in the Inmanta DSL. For example NullableType(Number()) represents number?.

class inmanta.ast.type.Primitive

Bases: Type

Abstract base class representing primitive types.

cast(value: object | None) → object

Cast a value to this type. If the value can not be cast, raises a inmanta.ast.RuntimeException.

class inmanta.ast.type.Number

Bases: Primitive

This class represents an integer or a float in the configuration model.

class inmanta.ast.type.Integer

Bases: Primitive

An instance of this class represents the int type in the configuration model.

class inmanta.ast.type.Bool

Bases: Primitive

This class represents a simple boolean that can hold true or false.

class inmanta.ast.type.String

Bases: Primitive

This class represents a string type in the configuration model.

class inmanta.ast.type.Union(types: Sequence[Type])

Bases: Type

Instances of this class represent a union of multiple types.

class inmanta.ast.type.Literal

Bases: Union

Instances of this class represent a literal in the configuration model. A literal is a primitive or a list or dict where all values are literals themselves.

class inmanta.ast.type.List

Bases: Type

Instances of this class represent a list type containing any types of values. This class refers to the list type used in plugin annotations. For the list type in the Inmanta DSL, see LiteralList.

class inmanta.ast.type.TypedList(element_type: Type)

Bases: List

Instances of this class represent a list type containing any values of type element_type. For example TypedList(Number()) represents number[].

class inmanta.ast.type.LiteralList

Bases: TypedList

Instances of this class represent a list type containing only Literal values. This is the list type in the DSL

class inmanta.ast.type.Dict

Bases: Type

Instances of this class represent a dict type with any types of values.

class inmanta.ast.type.TypedDict(element_type: Type)

Bases: Dict

Instances of this class represent a dict type containing only values of type element_type.

class inmanta.ast.type.LiteralDict

Bases: TypedDict

Instances of this class represent a dict type containing only Literal values. This is the dict type in the DSL

class inmanta.ast.type.ConstraintType(namespace: Namespace, name: str)

Bases: NamedType

A type that is based on a primitive type but defines additional constraints on this type. These constraints only apply on the value of the type.

inmanta.ast.type.TYPES

Maps Inmanta DSL types to their internal representation. For each key, value pair, value.type_string() is guaranteed to return key.

Note

The type classes themselves do not represent inmanta types, their instances do. For example, the type representation for the inmanta type number is Number(), not Number.

Protocol

class inmanta.protocol.common.Result(code: int = 0, result: dict[str, Any] | None = None, *, client: RESTClient | None = None, method_properties: MethodProperties[R] | None = None, environment: str | None = None)

A result of a method call

value() → R

Returns the value wrapped in this result, parsed as the method’s return type. Only works for typed methods. If paged, returns only the values for the page represented by this result. To get all results, see inmanta.protocol.common.PageableResult.all().

Converts return codes to http exceptions where applicable.

Raises:

BaseHttpException – when return code is not 200

class inmanta.protocol.common.PageableResult(code: int = 0, result: dict[str, Any] | None = None, *, client: RESTClient | None = None, method_properties: MethodProperties[R] | None = None, environment: str | None = None)

Bases: Result[list[V]], Generic[V]

Result for a list value, that offers methods for paging.

async all() → AsyncIterator[V]

Returns an async iterator over all values returned by this call. Follows paging links if there are any. Values are processed and validated as in value(), i.e. iterates over the value as returned by the API method, without wrapping in a Result object. If there are pages, simply chains results from multiple pages after each other.

all_sync(*, timeout: int = 120, ioloop: AbstractEventLoop | None = None) → Iterator[V]

Returns an iterator over all values returned by this call. For detailed behavior, see all().

In an async context, call all() instead.

Parameters:

timeout – The number of seconds to wait on each API call.

Data

Warning

In contrast to the rest of this section, the data API interface is subject to change. It is documented here because it is currently the only available API to interact with the data framework. A restructure of the data framework is expected at some point. Until then, this API should be considered unstable.

inmanta.data.TBaseDocument : typing.TypeVar

TypeVar with BaseDocument bound.

class inmanta.data.BaseDocument(from_postgres: bool = False, **kwargs: object)

A base document in the database. Subclasses of this document determine collections names. This type is mainly used to bundle query methods and generate validate and query methods for optimized DB access. This is not a full ODM.

Fields are modelled using type annotations similar to protocol and pydantic. The following is supported:

• Attributes are defined at class level with type annotations

• Attributes do not need a default value. When no default is provided, they are marked as required.

• When a value does not have to be set: either a default value or making it optional can be used. When a field is optional without a default value, none will be set as default value so that the field is available.

• Fields that should be ignored, can be added to __ignore_fields__ This attribute is a tuple of strings

• Fields that are part of the primary key should be added to the __primary_key__ attributes. This attribute is a tuple of strings.

async classmethod get_by_id(doc_id: UUID, connection: Connection | None = None) → TBaseDocument | None

Get a specific document based on its ID

Returns:

An instance of this class with its fields filled from the database.

async classmethod get_list(*, order_by_column: str | None = None, order: str | None = None, limit: int | None = None, offset: int | None = None, no_obj: bool | None = None, lock: RowLockMode | None = None, connection: Connection | None = None, **query: object) → list[TBaseDocument]

Get a list of documents matching the filter args

class inmanta.data.Compile(from_postgres: bool = False, **kwargs: object)

Bases: BaseDocument

A run of the compiler

Parameters:

• environment – The environment this resource is defined in

• requested – Time the compile was requested

• started – Time the compile started

• completed – Time to compile was completed

• do_export – should this compile perform an export

• force_update – should this compile definitely update

• metadata – exporter metadata to be passed to the compiler

• requested_environment_variables – environment variables requested to be passed to the compiler

• mergeable_environment_variables – environment variables to be passed to the compiler. These env vars can be compacted over multiple compiles. If multiple values are compacted, they will be joined using spaces.

• used_environment_variables – environment variables passed to the compiler, None before the compile is started

• success – was the compile successful

• handled – were all registered handlers executed?

• version – version exported by this compile

• remote_id – id as given by the requestor, used by the requestor to distinguish between different requests

• compile_data – json data as exported by compiling with the –export-compile-data parameter

• substitute_compile_id – id of this compile’s substitute compile, i.e. the compile request that is similar to this one that actually got compiled.

• partial – True if the compile only contains the entities/resources for the resource sets that should be updated

• removed_resource_sets – indicates the resource sets that should be removed from the model

• exporter_plugin – Specific exporter plugin to use

• notify_failed_compile – if true use the notification service to notify that a compile has failed. By default, notifications are enabled only for exporting compiles.

• failed_compile_message – Optional message to use when a notification for a failed compile is created

• soft_delete – Prevents deletion of resources in removed_resource_sets if they are being exported.

• links – An object that contains relevant links to this compile. It is a dictionary where the key is something that identifies one or more links and the value is a list of urls. i.e. {“instances”: [“link-1’,”link-2”], “compiles”: [“link-3”]}

async classmethod get_substitute_by_id(compile_id: UUID, connection: Connection | None = None) → Compile | None

Get a compile’s substitute compile if it exists, otherwise get the compile by id.

Parameters:

compile_id – The id of the compile for which to get the substitute compile.

Returns:

The compile object for compile c2 that is the substitute of compile c1 with the given id. If c1 does not have a substitute, returns c1 itself.

to_dto() → CompileRun

class inmanta.data.ConfigurationModel(**kwargs: object)

Bases: BaseDocument

A specific version of the configuration model.

Parameters:

• version – The version of the configuration model, represented by a unix timestamp.

• environment – The environment this configuration model is defined in

• date – The date this configuration model was created

• partial_base – If this version was calculated from a partial export, the version the partial was applied on.

• released – Is this model released and available for deployment?

• deployed – Is this model deployed?

• result – The result of the deployment. Success or error.

• version_info – Version metadata

• total – The total number of resources

• is_suitable_for_partial_compiles – This boolean indicates whether the model can later on be updated using a partial compile. In other words, the value is True iff no cross resource set dependencies exist between the resources.

async classmethod get_versions(environment: UUID, start: int = 0, limit: int = 100000, connection: Connection | None = None) → list[ConfigurationModel]

Get all versions for an environment ordered descending

class inmanta.data.Environment(from_postgres: bool = False, **kwargs: object)

Bases: BaseDocument

A deployment environment of a project

Parameters:

• id – A unique, machine generated id

• name – The name of the deployment environment.

• project – The project this environment belongs to.

• repo_url – The repository url that contains the configuration model code for this environment.

• repo_branch – The repository branch that contains the configuration model code for this environment.

• settings – Key/value settings for this environment. This dictionary does not necessarily contain a key for every environment setting known by the server.

• last_version – The last version number that was reserved for this environment

• description – The description of the environment

• icon – An icon for the environment

class inmanta.data.Report(from_postgres: bool = False, **kwargs: object)

Bases: BaseDocument

A report of a substep of compilation

Parameters:

• started – when the substep started

• completed – when it ended

• command – the command that was executed

• name – The name of this step

• errstream – what was reported on system err

• outstream – what was reported on system out

class inmanta.data.Resource(from_postgres: bool = False, **kwargs: object)

Bases: BaseDocument

A specific version of a resource. This entity contains the desired state of a resource.

Parameters:

• environment – The environment this resource version is defined in

• resource_id – The id of the resource (without the version)

• resource_type – The type of the resource

• resource_id_value – The attribute value from the resource id

• agent – The name of the agent responsible for deploying this resource

• attributes – The desired state for this version of the resource as a dict of attributes

• attribute_hash – hash of the attributes, excluding requires, provides and version, used to determine if a resource describes the same state across versions

• is_undefined – If the desired state for resource is undefined

• resource_set – The id of the resource set this resource belongs to.

async classmethod get_resources_for_version(environment: UUID, version: int, agent: str | None = None, no_obj: bool = False, *, connection: Connection | None = None) → list[Resource]

class inmanta.data.ResourceAction(from_postgres: bool = False, **kwargs: object)

Bases: BaseDocument

Log related to actions performed on a specific resource version by Inmanta.

Parameters:

• environment – The environment this action belongs to.

• version – The version of the configuration model this action belongs to.

• resource_version_ids – The resource version ids of the resources this action relates to.

• action_id – This id distinguishes the actions from each other. Action ids have to be unique per environment.

• action – The action performed on the resource

• started – When did the action start

• finished – When did the action finish

• messages – The log messages associated with this action

• status – The status of the resource when this action was finished

• changes – A dict with key the resource id and value a dict of fields -> value. Value is a dict that can contain old and current keys and the associated values. An empty dict indicates that the field was changed but not data was provided by the agent.

• change – The change result of an action

async classmethod get_logs_for_version(environment: UUID, version: int, action: str | None = None, limit: int = 0, connection: Connection | None = None) → list[ResourceAction]

class inmanta.data.model.BaseModel

Bases: DateTimeNormalizerModel

Base class for all data objects in Inmanta

inmanta.data.model.ResourceIdStr = inmanta.types.ResourceIdStr

NewType creates simple unique types with almost zero runtime overhead.

NewType(name, tp) is considered a subtype of tp by static type checkers. At runtime, NewType(name, tp) returns a dummy callable that simply returns its argument.

Usage:

UserId = NewType('UserId', int)

def name_by_id(user_id: UserId) -> str:
    ...

UserId('user')          # Fails type check

name_by_id(42)          # Fails type check
name_by_id(UserId(42))  # OK

num = UserId(5) + 1     # type: int

inmanta.data.model.ResourceVersionIdStr = inmanta.types.ResourceVersionIdStr

NewType creates simple unique types with almost zero runtime overhead.

NewType(name, tp) is considered a subtype of tp by static type checkers. At runtime, NewType(name, tp) returns a dummy callable that simply returns its argument.

Usage:

UserId = NewType('UserId', int)

def name_by_id(user_id: UserId) -> str:
    ...

UserId('user')          # Fails type check

name_by_id(42)          # Fails type check
name_by_id(UserId(42))  # OK

num = UserId(5) + 1     # type: int

class inmanta.db.util.PGRestore(script: list[str], postgresql_client: Connection)

Bases: object

Class that offers support to restore a database dump.

This class assumes that the names of schemas, tables and columns in the dump don’t contain a dot, double quote or whitespace character.

async inmanta.db.util.clear_database(postgresql_client: Connection) → None

Remove all content from the database. Removes functions, tables and data types.

Domain conversion

This section describes methods for converting values between the plugin domain and the internal domain. This conversion is performed automatically for plugin arguments and return values so it is only required when bypassing the usual plugin workflow by calling internal methods directly.

class inmanta.execute.proxy.DynamicProxy

This class wraps an object and makes sure that a model is never modified by native code.

classmethod return_value(value: object, *, context: ProxyContext | None = None) → None | str | tuple[object, ...] | int | float | bool | DynamicProxy | Reference

Converts a value from the internal domain to the plugin domain.

Parameters:

context – The context in which the given object lives. If None, assumes that the value has already passed validation. When None, the object’s string representation, rather than its name is used for error reporting.

classmethod unwrap(item: object, *, dynamic_context: DynamicUnwrapContext | None = None) → object

Converts a value from the plugin domain to the internal domain.

Parameters:

dynamic_context – a type resolver context. When passed in, dataclasses are converted as well.

Server

class inmanta.server.extensions.ApplicationContext

Bases: object

class inmanta.server.bootloader.InmantaBootloader(configure_logging: bool = False)

Bases: object

The inmanta bootloader is responsible for: - discovering extensions - loading extensions - loading core and extension slices - starting the server and its slices in the correct order