Library Service

The ASAB Library is the concept of shared data content across microservices in the cluster. In the cluster/cloud microservice architectures, all microservices must have access to unified resources. The Library provides a read-only interface for listing and reading this content.

The Library is designed to be read-only. It also allows to "stack" various libraries into one view (overlayed), merging the content of each library into one united space.

The library can also notify the ASAB microservice about changes, e.g. for automated update/reload.

Library structure

The library content is organized in a simplified file system manner, with directories and files.

Example of the library structure

+ /MyDirectory/
    - /MyDirectory/item1.yaml
    - /MyDirectory/item2.json
+ /AnotherDirectory/
    - /AnotherDirectory/item3.yaml
    + /AnotherDirectory/Folder/
        - /folder2/Folder/item4.json

Library path rules

• Any path must start with "/", including the root path.
• The directory path must end with "/".
• The directory name cannot contain ".".
• The item path must end with a file extension (e.g. ".txt", ".json", ...).

Layers

The library content can be organized into an unlimited number of layers. Each layer is represented by a provider (e.g. filesystem, zookeeper, git, ...) with a specific configuration. Two layers can have the same provider but different base paths.

The layers of the library are like slices of Swiss cheese layered on top of each other. Only if there is a hole in the top layer can you see the layer that shows through underneath. It means that files of the upper layer overwrite files with the same path in the lower layers.

Illustration of ASAB Library layers. Green items are visible, grey items are hidden. Moreover, Layer 1 contains .disabled.yaml file containing the list of disabled files.

Tip

In the most applications, it is common to create the first layer with Zookeeper provider and the layers beneath with git or libsreg provider. This allows you to quickly modify items of the Library on the first layer.

Disabling files

The library concept supports multi-tenancy. By default, all items of the library are visible for everyone, but you can disable some of them for specific tenants.

In order to disable some items of the library, create a file /.disabled.yaml. This file must be created on the first layer.

In the following example, file1.txt is disabled for tenant-1 and tenant-2, file2.txt for tenant-1 and file3.txt for every tenant.

.disabled.yaml

/file1.txt:
    - tenant-1
    - tenant-2
/file2.txt: tenant-1
/file3.txt: '*'

Warning

When disabling a file for all tenants with a star, don't forget to close it in quotation marks. Otherwise, YAML would interpret star as an alias. Read more about anchors and aliases.

Library service

The library service may exist in multiple instances, with different paths setups. For that reason, you have to provide a unique service_name and there is no default value for that.

Each Library item is represented by LibraryItem dataclass. Read more in the reference section.

Example of the use:

import asab
import asab.library

class MyApplication(asab.Application):

    async def initialize(self):
        self.LibraryService = asab.library.LibraryService(self, "LibraryService") #(1)!
        self.PubSub.subscribe("Library.ready!", self.on_library_ready) #(2)!

    async def on_library_ready(self, event_name, library): #(3)!

        for item in await self.LibraryService.list("/", recursive=True): #(4)!
            print("*", item)
            if item.type == 'item': #(5)!
                itemio = await self.LibraryService.read(item.name) #(6)!
                if itemio is not None:
                    with itemio: #(7)!
                        content = itemio.read()
                        print("- content: {} bytes".format(len(content)))
                else:
                    print("  - (DISABLED)")

if __name__ == '__main__':
    app = MyApplication()
    app.run()

1. Initializes the Library Service. Remember to specify a unique service_name.
2. When the Library is initialized, Library.ready! PubSub message is emitted.
3. The callback has to possess two arguments. event_name is the message "Library.ready!", library is the specific provider with which is the Library initialized.
4. list() method returns list of LibraryItems. For more information, see the reference section.
5. item.type can be either 'item' or 'dir'.
6. read() coroutine returns item IO object or None if the file is disabled.
7. Item IO object is used as a context manager.

Example of the library configuration:

[library]
providers:
    provider+1://...
    provider+2://...
    provider+3://...

PubSub messages

The Library is created in not-ready state. After the connection with the technologies behind is established, every library provider changes its state to ready. The Library switches to ready state after all its providers are ready.

If some of the providers is disconnected, the Library switches to not-ready state again till the connection is reestablished.

Every time the Library changes its state, PubSub message is published, with the arguments provider and path.

Message	Published when...
Library.not_ready!	at least one provider is not ready.
Library.ready!	all of the providers are ready.
Library.change!	the content of the Library has changed.

Notification on changes

Some providers are able to detect changes of the library items.

Example

class MyApplication(asab.Application):

async def initialize(self):
    self.PubSub.subscribe("Library.ready!", self.on_library_ready
    self.PubSub.subscribe("Library.change!", self.on_library_change)

async def on_library_ready(self, event_name, library=None):
    await self.LibraryService.subscribe(["/asab"]) #(1)!

def on_library_change(self, message, provider, path): #(2)!
    print("New changes in the library found by provider: '{}'".format(provider))

1. self.LibraryService.subscribe() method takes either a single path as a string or multiple paths in list and watches for changes in them.
2. This coroutine takes three arguments: message (Library.change! in this case), provider (name of the provider that has detected changes) and path (the path where changes were made).

Info

Note that the some of the providers detect changes immediately while others detect them periodically. For example, git provider pulls the repository every minute, only after that the changes can be detected.

Providers

The list of available providers:

Provider	Read the content	Notify on changes
Filesystem
Apache Zookeeper
Microsoft Azure Storage
Git
Libraries repository

Filesystem

The most basic provider that reads data from the local filesystem. The notification on changes functionality is available only for Linux systems, as it uses inotify.

Configuration examples:

[library]
providers: /home/user/directory

[library]
providers: ./this_directory

[library]
providers: file:///home/user/directory

Apache Zookeeper

ZooKeeper as a consensus technology is vital for microservices in the cluster.

There are several configuration strategies:

1) Configuration from [zookeeper] section.

[zookeeper]
servers=zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
path=/library

[library]
providers:
    zk://

2) Specify a path of a ZooKeeper node where only library lives.

[zookeeper]
servers=zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
path=/else

[library]
providers:
    zk:///library

[zookeeper]
servers=zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
path=/else

[library]
providers:
    zk:///

3) Configuration from the URL in the [library] section.

[library]
providers:
    zk://zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181/library

4) Configuration from [zookeeper] section and joined [path]{.title-ref} from [zookeeper] and [library] sections.

> The resulting path will be [/else/library]{.title-ref}.

[zookeeper]
servers=zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
path=/else

[library]
providers:
    zk://./library

If a path from the [zookeeper] section is missing, an application class name will be used, e.g. /BSQueryApp/library.

Working with target=tenant Layers

The ASAB Library supports multi-tenancy, allowing you to manage and separate content specific to different tenants within layers in library. To handle tenant-specific contexts, use Tenant.set() to set the context and Tenant.reset() to reset it after processing.

Implementing Tenant-Specific Logic in Your Application

To handle tenant-specific logic, make sure the AuthService is present in app.py:

# Install the AuthService for tenant-based authentication
self.AuthService = asab.web.auth.AuthService(self)
self.AuthService.install(self.WebContainer)

Example: Processing Multiple Tenants

The following example demonstrates how to iterate through multiple tenants, setting and resetting the tenant context for each one:

import asab.contextvars

async def process_tenants(self):
    for tenant in self.Tenants:
        # Set the tenant context
        tenant_context = asab.contextvars.Tenant.set(tenant)
        try:
            # Process tenant-specific logic here
            print(f"Processing workflows for tenant: {tenant}")
        finally:
            # Reset the tenant context
            asab.contextvars.Tenant.reset(tenant_context)

In this method: - Tenant.set(tenant) establishes the context for the current tenant. - Tenant.reset(tenant_context) ensures the context is cleared after processing, preventing any unintended carryover.

Example: Handling a Single Tenant

If you need to handle only one tenant context, the process is straightforward:

import asab.contextvars

async def process_single_tenant(self, tenant):
    tenant_context = asab.contextvars.Tenant.set(tenant)
    try:
        # Perform operations for the tenant
        print(f"Processing data for tenant: {tenant}")
    finally:
        asab.contextvars.Tenant.reset(tenant_context)

Microsoft Azure Storage

You can configure the microservice to read from the Microsoft Azure Storage container.

Configuration:

[library]
providers: azure+https://ACCOUNT-NAME.blob.core.windows.net/BLOB-CONTAINER

If Container Public Access Level is not set to "Public access", then "Access Policy" must be created with "Read" and "List" permissions and "Shared Access Signature" (SAS) query string must be added to a URL in a configuration:

[library]
providers: azure+https://ACCOUNT-NAME.blob.core.windows.net/BLOB-CONTAINER?sv=2020-10-02&si=XXXX&sr=c&sig=XXXXXXXXXXXXXX

Git repository

Warning

Connection to git repositories requires pygit2 library to be installed.

pip install pygit2

Please follow this format in the configuration:

[library]
providers: git+http(s)://<username>:<deploy-token>@<path>#<branch>

Cloning from GitHub repository:

Using a public repository from GitHub, the configuration may look like this:

[library]
providers: git+https://github.com/john/awesome_project.git

Using custom branch:

Use hash #<branch-name> to clone a repository from a selected branch:

[library]
providers: git+https://github.com/john/awesome_project.git#name-of-the-branch

Deploy tokens in GitLab

GitLab uses deploy tokens to enable authentication of deployment tasks, independent of a user account. Authentication through deploy tokens is the only supported option for now.

If you want to create a deploy token for your GitLab repository, follow these steps from the manual:

1. Go to Settings > Repository > Deploy tokens section in your repository. (Note that you have to possess a "Maintainer" or "Owner" role for the repository.)
2. Expand the "Deploy tokens" section. The list of current Active Deploy Tokens will be displayed.
3. Complete the fields and scopes. We recommend a custom "username", as you will need it later for the URL in the configuration.
4. Record the deploy token's values before leaving or refreshing the page! After that, you cannot access it again.

After the deploy token is created, use the URL for the repository in the following format:

[library]
providers: git+https://<username>:<deploy_token>@gitlab.example.com/john/awesome_project.git

Where does the repository clone?

The git provider clones the repository into a temporary directory. The default path for the cloned repository is /tmp/asab.library.git/ and it can be changed manually:

[library:git]
repodir=path/to/repository/cache

Libraries repository

The libsreg provider downloads the content from the distribution URL. The distribution URL points to HTTP(S) server where content archives are published.

Configuration examples:

[library]
providers: libsreg+https://libsreg.example.com/my-library

More than one distribution server can be specified:

[library]
providers: libsreg+https://libsreg1.example.com,libsreg2.example.com/my-library

This variant provides more resiliency against a distribution server unavailability.

A structure of the distribution server filesystem:

+ /my-library/
  - my-library-master.tar.xz
  - my-library-master.tar.xz.sha256
  - my-library-production.tar.xz
  - my-library-production.tar.xz.sha256
  - my-library-v43.41.tar.xz
  - my-library-v43.41.tar.xz.sha256
  ...

• *.tar.xz: This is the TAR/XZ archive of the actual content
• *.tar.xz.sha256: SHA256 checksum of the archive

The structure of the distribution is as follows:

/{archname}/{archname}-{version}.tar.xz

• archname: A name of the distribution archive, my-library in the example above
• version: A version of the distribution archive, master, production are typically GIT branches, v43.41 is a GIT tag.

Tip

This provider is designed to use Microsoft Azure Storage as a distribution point. Is is assumed that the content archives are uploaded to the distribution point using CI/CD.

Reference

asab.library.LibraryService

Bases: Service

Configuration:

[library]
providers:
        provider+1://
        provider+2://
        provider+3://

The order of providers is important, the priority (or layering) is top-down.

Each library provider is specified by URL/URI schema:

• zk:// or zookeeper:// for ZooKeeper provider
• file:// or local path for FileSystem provider
• azure+https:// for Microsoft Azure Storage provider.
• git+https:// for Git provider.
• libsreg+https:// for Libraries provider.

The first provider is responsible for providing /.disabled.yaml.

A library is created in “not ready” state, each provider then informs the library when it is ready (eg. Zookeeper provider needs to connect to Zookeeper servers). Only after all providers are ready, the library itself becomes ready. The library indicates that by the PubSub event Library.ready!.

Source code in asab/library/service.py

class LibraryService(Service):
	"""
	Configuration:

	```ini
	[library]
	providers:
		provider+1://
		provider+2://
		provider+3://
	```

	The order of providers is important, the priority (or layering) is top-down.

	Each library provider is specified by URL/URI schema:

	* `zk://` or `zookeeper://` for ZooKeeper provider
	* `file://` or local path for FileSystem provider
	* `azure+https://` for Microsoft Azure Storage provider.
	* `git+https://` for Git provider.
	* `libsreg+https://` for Libraries provider.

	The first provider is responsible for providing `/.disabled.yaml`.

	A library is created in “not ready” state, each provider then informs the library when it is ready
	(eg. Zookeeper provider needs to connect to Zookeeper servers). Only after all providers are ready, the library itself becomes ready.
	The library indicates that by the PubSub event `Library.ready!`.
	"""

	def __init__(
		self,
		app: Application,
		service_name: str,
		paths: typing.Union[str, typing.List[str], None] = None
	):
		"""
		Initialize the LibraryService.

		The library service is designed to "exist" in multiple instances,
		with different `paths` setups.
		For that reason, you have to provide unique `service_name`
		and there is no _default_ value for that.

		If `paths` are not provided, they are fetched from `[library]providers` configuration.

		Args:
			app: The ASAB Application.
			service_name: A unique name of the service.
			paths (str | list[str] | None ): Either single path or list of paths with which LibraryService is connected.
		"""

		super().__init__(app, service_name)
		self.Libraries: list[LibraryProviderABC] = []
		self.Disabled: dict = {}
		self.DisabledPaths: list = []
		self.Favorites: dict = {}
		self.FavoritePaths: list = []


		if paths is None:
			# load them from configuration
			try:
				paths = Config.getmultiline("library", "providers")
			except configparser.NoOptionError:
				L.critical("'providers' option is not present in configuration section 'library'.")
				raise SystemExit("Exit due to a critical configuration error.")

		# paths can be string if specified as argument
		if isinstance(paths, str):
			paths = re.split(r"\s+", paths)

		for layer, path in enumerate(paths):
			# Create library for each layer of paths
			self._create_library(path, layer)

		app.PubSub.subscribe("Application.tick/60!", self._on_tick60)


	async def finalize(self, app):
		while len(self.Libraries) > 0:
			lib = self.Libraries.pop(-1)
			await lib.finalize(self.App)

	async def _on_tick60(self, message_type):
		await self._read_disabled()
		await self._read_favorites()

	def _create_library(self, path, layer):
		library_provider = None
		if path.startswith('zk://') or path.startswith('zookeeper://'):
			from .providers.zookeeper import ZooKeeperLibraryProvider
			library_provider = ZooKeeperLibraryProvider(self, path, layer)

		elif path.startswith('./') or path.startswith('/') or path.startswith('file://'):
			from .providers.filesystem import FileSystemLibraryProvider
			library_provider = FileSystemLibraryProvider(self, path, layer)

		elif path.startswith('azure+https://'):
			from .providers.azurestorage import AzureStorageLibraryProvider
			library_provider = AzureStorageLibraryProvider(self, path, layer)

		elif path.startswith('git+'):
			from .providers.git import GitLibraryProvider
			library_provider = GitLibraryProvider(self, path, layer)

		elif path.startswith('libsreg+'):
			from .providers.libsreg import LibsRegLibraryProvider
			library_provider = LibsRegLibraryProvider(self, path, layer)

		elif path == '' or path.startswith("#") or path.startswith(";"):
			# This is empty or commented line
			return

		else:
			L.error("Incorrect/unknown provider for '{}'".format(path))
			raise SystemExit("Exit due to a critical configuration error.")

		self.Libraries.append(library_provider)

	def is_ready(self) -> bool:
		"""
		Check if all the library providers are ready.

		Returns:
			True if every provider is ready; if even one provider is not, returns False.
		"""
		if not self.Libraries:
			return False

		for provider in self.Libraries:
			if not provider.IsReady:
				return False
		return True


	async def _set_ready(self, provider):
		if len(self.Libraries) == 0:
			return

		if (provider == self.Libraries[0]) and provider.IsReady:
			await self._read_disabled()
			await self._read_favorites()

		if self.is_ready():
			L.log(LOG_NOTICE, "is ready.", struct_data={'name': self.Name})
			self.App.PubSub.publish("Library.ready!", self)
		elif not provider.IsReady:
			L.log(LOG_NOTICE, "is NOT ready.", struct_data={'name': self.Name})
			self.App.PubSub.publish("Library.not_ready!", self)

	def _ensure_ready(self):
		if not self.is_ready():
			raise LibraryNotReadyError("Library is not ready yet.")

	async def find(self, path: str) -> typing.List[str]:
		"""
		Search for files with a specific name within a library, using the provided path.

		The method traverses the library directories, looking for files that match the given filename.
		It returns a list of paths leading to these files, empty if no items are found.

		Args:
			path (str): Location of the file in Library. It must start with a forward slash and include the filename. Example: '/Dashboards/Cisco/Overview.json'

		Returns:
			typing.List[str]: A list of paths to the found files. If no files are found, the list will be empty.
		"""
		_validate_path_item(path)

		results = []
		for library in self.Libraries:
			found_files = await library.find(path)
			if found_files:
				results.extend(found_files)
		return results

	async def read(self, path: str) -> typing.Optional[typing.IO]:
		"""
		THIS IS OBSOLETED METHOD, USE `open(...)` !!!

		Read the content of the library item specified by `path`. This method can be used only after the Library is ready.

		Args:
			path (str): Path to the file, `LibraryItem.name` can be used directly.
			tenant (str | None): The tenant to apply. If not specified, the global access is assumed.

		Returns:
			( IO | None ): Readable stream with the content of the library item. `None` is returned if the item is not found or if it is disabled (either globally or for the specified tenant).

		Example:

		```python
		itemio = await library.read('/path', 'tenant')
		if itemio is not None:
			with itemio:
				return itemio.read()
		```
		"""
		self._ensure_ready()

		LogObsolete.warning("Method 'LibraryService.read()' is obsolete. Use 'LibraryService.open()' method instead.")
		_validate_path_item(path)

		if self.check_disabled(path):
			return None

		for library in self.Libraries:
			itemio = await library.read(path)
			if itemio is None:
				continue
			return itemio

		return None


	@contextlib.asynccontextmanager
	async def open(self, path: str):
		"""
		Read the content of the library item specified by `path` in a SAFE way, protected by a context manager/with statement.
		This method can be used only after the Library is ready.

		Example:

		```python
		async with self.LibraryService.open(path) as io:
			if io is None:
				return None
			text = b.read().decode("utf-8")
		```
		"""
		self._ensure_ready()

		_validate_path_item(path)

		# Same functionality as in read() method
		itemio = None
		disabled = self.check_disabled(path)
		if not disabled:
			for library in self.Libraries:
				itemio = await library.read(path)
				if itemio is not None:
					break

		if itemio is None:
			yield itemio
		else:
			try:
				yield itemio
			finally:
				itemio.close()


	async def list(self, path: str = "/", recursive: bool = False) -> typing.List[LibraryItem]:
		"""
		List the directory of the library specified by the path that are enabled for the specified tenant.
		This method can be used only after the Library is ready.

		**WARNING:** Tenant must be set in the context variable!
		If it is not set automatically (e.g. from web request), it must be set manually.

		Example:

		```python
		try:
			tenant_ctx = asab.contextvars.Tenant.set(tenant)
			items = self.LibraryService.list(path)
			...
		finally:
			asab.contextvars.Tenant.reset(tenant_ctx)
		```

		Args:
			path (str): Path to the directory.
			recursive (bool): If `True`, return a list of items located at `path` and its subdirectories.

		Returns:
			List of items that are enabled for the tenant.
		"""
		self._ensure_ready()

		_validate_path_directory(path)

		# List requested level using all available providers
		items = await self._list(path, providers=self.Libraries)

		if recursive:
			# If recursive scan is requested, then iterate thru list of items
			# find 'dir' types there and list them.
			# Output of this list is attached to the list for recursive scan
			# and also to the final output
			recitems = list(items[:])

			while len(recitems) > 0:

				item = recitems.pop(0)
				if item.type != 'dir':
					continue

				child_items = await self._list(item.name, providers=item.providers)
				items.extend(child_items)
				recitems.extend(child_items)

		return items

	async def _list(self, path, providers):
		"""
		Lists items from all providers, merging items with the same name,
		and simply adding every layer an item belongs to.

		Args:
			path (str): The path to list items from.
			providers (list): A list of providers to query.

		Returns:
			list: A list of unique LibraryItem objects with merged layers.
		"""
		items: list[LibraryItem] = []
		unique_items: dict[str, LibraryItem] = {}

		# Launch tasks to list items from each provider.
		tasks = [(self.Libraries.index(provider), asyncio.create_task(provider.list(path))) for provider in providers]

		for outer_layer, task in tasks:
			try:
				provider_items: list[LibraryItem] = await task
			except KeyError:
				# The path doesn't exist in this provider.
				continue
			except Exception:
				L.exception("Unexpected error when listing path '{}' on layer {}.".format(path, outer_layer))
				continue

			for item in provider_items:
				# Check if the item is disabled.
				item.disabled = self.check_disabled(item.name)
				item.favorite = self.check_favorite(item.name)
				# Use the layers as provided by the provider; if empty, fall back to outer_layer.
				provider_layers = item.layers if item.layers else [outer_layer]

				if item.name in unique_items:
					existing_item = unique_items[item.name]
					# Merge providers for directories (avoid duplicates).
					if existing_item.type == "dir" and item.type == "dir":
						for p in item.providers:
							if p not in existing_item.providers:
								existing_item.providers.append(p)
					# Extend the layers list by adding each new layer (without sorting).
					for layer_value in provider_layers:
						if layer_value not in existing_item.layers:
							existing_item.layers.append(layer_value)
				else:
					# New item: use the provided layers.
					item.layers = provider_layers
					unique_items[item.name] = item
					items.append(item)

		# Do not sort items; return them in the order they were merged.
		return items


	async def _read_favorites(self):
		"""
		Load favorites from '/.favorites.yaml' into:
			- self.Favorites: { '/path/file.ext': ['tenant', '*', ...] }
			- self.FavoritePaths: [ ('/path/folder/', ['tenant', '*', ...]), ... ]
		Expected YAML shape:
			/path:
				tenants:
				- system
		"""
		fav_data = None

		try:
			fav_file = await self.Libraries[0].read('/.favorites.yaml')
		except Exception as e:
			L.warning("Failed to read '/.favorites.yaml': {}.".format(e))
			self.Favorites = {}
			self.FavoritePaths = []
			return

		if fav_file is None:
			self.Favorites = {}
			self.FavoritePaths = []
			return

		try:
			fav_data = yaml.load(fav_file, Loader=yaml.CSafeLoader)
		except Exception:
			L.exception("Failed to parse '/.favorites.yaml'")
			self.Favorites = {}
			self.FavoritePaths = []
			return
		finally:
			if hasattr(fav_file, "close"):
				try:
					fav_file.close()
				except Exception:
					pass

		if fav_data is None:
			self.Favorites = {}
			self.FavoritePaths = []
			return

		if not isinstance(fav_data, dict):
			L.warning("Unexpected favorites format ({}). Resetting.".format(type(fav_data).__name__))
			self.Favorites = {}
			self.FavoritePaths = []
			return

		files = {}
		folders = []
		for k, v in fav_data.items():
			if not isinstance(k, str):
				L.warning("Ignoring non-string favorite key: {}".format(k))
				continue

			# normalize tenants list
			tenants = []
			if isinstance(v, dict) and 'tenants' in v:
				tv = v.get('tenants')
				if isinstance(tv, list):
					tenants = [str(t) for t in tv]
				elif isinstance(tv, set):
					tenants = [str(t) for t in list(tv)]
				elif tv is None:
					tenants = []
				else:
					tenants = [str(tv)]
			elif isinstance(v, list):
				# Back-compat: allow direct list
				tenants = [str(t) for t in v]
			else:
				tenants = [str(v)]

			if k.endswith('/'):
				folders.append((k, tenants))
			else:
				files[k] = tenants

		# Sort folders shortest→longest (consistent with DisabledPaths sort)
		folders.sort(key=lambda x: len(x[0]))

		self.Favorites = files
		self.FavoritePaths = folders

	async def _read_disabled(self, publish_changes=False):

		old_disabled = self.Disabled.copy()
		old_disabled_paths = list(self.DisabledPaths)
		# Read the file
		disabled_file = await self.Libraries[0].read('/.disabled.yaml')

		if disabled_file is None:
			self.Disabled = {}
			self.DisabledPaths = []
		else:
			try:
				disabled_data = yaml.load(disabled_file, Loader=yaml.CSafeLoader)
			except Exception:
				L.exception("Failed to parse '/.disabled.yaml'")
				self.Disabled = {}
				self.DisabledPaths = []
				return

			if disabled_data is None:
				self.Disabled = {}
				self.DisabledPaths = []
				return

			if isinstance(disabled_data, set):
				# Backward compatibility (August 2023)
				self.Disabled = {key: '*' for key in disabled_data}
				self.DisabledPaths = []
			else:
				self.Disabled = {}
				self.DisabledPaths = []
				for k, v in disabled_data.items():
					if k.endswith('/'):
						self.DisabledPaths.append((k, v))
					else:
						self.Disabled[k] = v

		self.DisabledPaths.sort(key=lambda x: len(x[0]))

		# If requested, compare old and new disables to notify subscribers via Library.change!
		if publish_changes:
			await self._publish_change_for_disabled_diff(old_disabled, old_disabled_paths)

	async def _publish_change_for_disabled_diff(self, old_disabled, old_disabled_paths):
		"""
		Compare old and new disabled data and publish Library.change!
		if any subscribed path is affected for the specific subscription target.
		"""

		if not self.Libraries:
			return

		# Only the first provider (layer 0) owns and manages '/.disabled.yaml',
		# so all disable‐and‐subscription logic is driven from that topmost layer.
		provider = self.Libraries[0]

		# Check if the provider has Subscriptions attribute
		subscriptions = getattr(provider, "Subscriptions", None)
		if subscriptions is None:
			return

		# For each subscription (path + target)
		for p_target, p_path in list(subscriptions):
			# Check if something disabled under this path and target changed
			changed = self._is_disabled_diff_affecting_path(p_path, old_disabled, old_disabled_paths, p_target)

			if changed:
				self.App.PubSub.publish("Library.change!", self, p_path)

	def _is_disabled_diff_affecting_path(self, sub_path, old_disabled, old_disabled_paths, target=None):
		"""
		Check if disabling changes affect the subscribed path for a specific target.
		For target=="tenant" (wildcard), we fire whenever the set of tenants changes.
		"""

		# Normalize path (ensure it ends with / for folders)
		if not sub_path.endswith('/'):
			sub_path = sub_path + '/'

		# 1) File-level disables
		for path in old_disabled.keys() | self.Disabled.keys():
			if not path.startswith(sub_path):
				continue

			old_entry = old_disabled.get(path) or []
			new_entry = self.Disabled.get(path) or []

			if target == "tenant":
				# Wildcard tenant subscription: any change in the list of tenants
				if set(old_entry) != set(new_entry):
					return True
			else:
				old_flag = self._is_disabled_for_target(old_entry, target)
				new_flag = self._is_disabled_for_target(new_entry, target)
				if old_flag != new_flag:
					return True

		# 2) Folder-level disables
		old_map = {p: v for p, v in old_disabled_paths}
		new_map = {p: v for p, v in self.DisabledPaths}

		for path in old_map.keys() | new_map.keys():
			if not path.startswith(sub_path):
				continue

			old_entry = old_map.get(path) or []
			new_entry = new_map.get(path) or []

			if target == "tenant":
				if set(old_entry) != set(new_entry):
					return True
			else:
				old_flag = self._is_disabled_for_target(old_entry, target)
				new_flag = self._is_disabled_for_target(new_entry, target)
				if old_flag != new_flag:
					return True

		return False

	def _is_disabled_for_target(self, disabled_entry, target):
		"""
		Check if a disabled entry affects a specific target (global, tenant, or tenant ID).
		"""
		if disabled_entry is None:
			return False

		if target is None or target == "global":
			return "*" in disabled_entry

		if target == "tenant":
			# Wildcard tenant subscription (all tenants)
			return bool(disabled_entry)

		if isinstance(target, tuple) and target[0] == "tenant":
			tenant_id = target[1]
			return "*" in disabled_entry or tenant_id in disabled_entry

		return False


	def check_disabled(self, path: str) -> bool:
		"""
		Check if the item specified in path is disabled, either globally or for the specified tenant.

		**WARNING:** When checking for items disabled for a tenant,
		it must be set in context variable before using this function!
		If it is not set automatically (e.g. from web request), it must be set manually.

		Example:

		1. Is path disabled for a specific tenant?

			```python
			try:
				tenant_ctx = asab.contextvars.Tenant.set(tenant)
				disabled = self.LibraryService.check_disabled(path)
				...
			finally:
				asab.contextvars.Tenant.reset(tenant_ctx)
			```

		2. Is path disabled globally?

			```python
			disabled = self.LibraryService.check_disabled(path)
			```

		Args:
			path (str): Path to the item to be checked.

		Returns:
			`True` if the item is disabled for the tenant.
		"""
		if not isinstance(path, str) or not path:
			raise LibraryInvalidPathError(
				message="Argument 'path' must be a non-empty string.",
				path=path,
			)
		try:
			tenant = Tenant.get()
		except LookupError:
			tenant = None

		# First check disabled by path
		for dp, disabled in self.DisabledPaths:
			if path.startswith(dp):
				if '*' in disabled:
					# Path is disabled for everybody
					return True

				if tenant is not None and tenant in disabled:
					# Path is disabled for a specified tenant
					return True

		# Then check for a specific item entries

		disabled = self.Disabled.get(path)

		if disabled is None:
			return False

		if '*' in disabled:
			# Item is disabled for everybody
			return True

		if tenant is not None and tenant in disabled:
			# Item is disabled for a specified tenant
			return True

		return False

	def check_favorite(self, path: str, inherit: bool = False) -> bool:
		"""
		Check if `path` is marked as favorite for the current tenant (or globally via '*').

		If `inherit` is True and a parent *folder* is favorited for the tenant/'*',
		any child path under that folder is considered favorited.

		**WARNING:** Tenant must be set in the context variable before using this function
		if you want tenant-aware behavior (same as check_disabled).

		Examples:

		1) Tenant-aware check:

			```python
			try:
				tenant_ctx = asab.contextvars.Tenant.set(tenant)
				is_fav = self.LibraryService.check_favorite(path)
				...
			finally:
				asab.contextvars.Tenant.reset(tenant_ctx)
			```

		2) Global-only check (any tenant):

			```python
			is_fav = self.LibraryService.check_favorite(path)
			```

		Args:
			path (str): Path to the item or folder to be checked.
			inherit (bool): If True, a favorited folder marks all descendants as favorited.

		Returns:
			bool: True if favorited for current tenant or globally.
		"""
		if not isinstance(path, str) or not path:
			raise LibraryInvalidPathError(
				message="Argument 'path' must be a non-empty string.",
				path=path,
			)

		try:
			tenant = Tenant.get()
		except LookupError:
			tenant = None

		# 1) Folder favorites
		#    - Exact match if not inheriting
		#    - Prefix match if inheriting (folder favorite applies to children)
		if inherit:
			for fp, fav_tenants in self.FavoritePaths:
				if path.startswith(fp):
					if '*' in fav_tenants:
						return True
					if tenant is not None and tenant in fav_tenants:
						return True
		else:
			# exact folder favorite (no inheritance)
			for fp, fav_tenants in self.FavoritePaths:
				if path == fp:
					if '*' in fav_tenants:
						return True
					if tenant is not None and tenant in fav_tenants:
						return True

		# 2) Exact item favorites
		fav_tenants = self.Favorites.get(path)
		if fav_tenants is None:
			return False

		if '*' in fav_tenants:
			return True

		if tenant is not None and tenant in fav_tenants:
			return True

		return False

	async def get_item_metadata(self, path: str) -> typing.Optional[dict]:
		"""
		Retrieve metadata for a specific file in the library, including its `target`.

		Args:
			path (str): The absolute path of the file to retrieve metadata for.
						Must start with '/' and include a filename with an extension.

		Returns:
			dict: Metadata for the specified file, including `target`, or None if not found.
		"""

		# Validate the path format
		_validate_path_item(path)

		# Split into directory and filename
		directory, filename = os.path.split(path)

		if not directory or not filename:
			L.warning("Invalid path '{}': missing directory or filename.".format(path))
			return None
		# Ensure directory ends with '/'
		if not directory.endswith('/'):
			directory += '/'

		try:
			# Fetch all items in the directory
			items = await self.list(directory)
		except Exception as e:
			L.warning("Failed to list items in directory '{}': {}".format(directory, e))
			return None

		# Use dictionary for faster lookup
		items_dict = {item.name: item for item in items}

		# Retrieve the item by path
		item = items_dict.get(path)
		if item and item.type == "item":
			# Match found; return metadata including `target`
			return {
				"name": item.name,
				"type": item.type,
				"layers": item.layers,
				"providers": item.providers,
				"disabled": item.disabled,
				"favorite": item.favorite,
				"override": item.override,
			}

		# Item not found
		L.info("Item '{}' not found in directory '{}'.".format(filename, directory))
		return None


	async def export(self, path: str = "/", remove_path: bool = False) -> typing.IO:
		"""
		Return a file-like stream containing a gzipped tar archive of the library contents of the path.

		Args:
			path: The path to export.
			tenant (str | None ): The tenant to use for the operation.
			remove_path: If `True`, the path will be removed from the tar file.

		Returns:
			A file object containing a gzipped tar archive.
		"""
		self._ensure_ready()

		_validate_path_directory(path)

		fileobj = tempfile.TemporaryFile()
		tarobj = tarfile.open(name=None, mode='w:gz', fileobj=fileobj)

		items = await self._list(path, providers=self.Libraries[:1])
		recitems = list(items[:])

		while len(recitems) > 0:

			item = recitems.pop(0)
			if item.type != 'dir':
				continue

			child_items = await self._list(item.name, providers=item.providers)
			items.extend(child_items)
			recitems.extend(child_items)

		for item in items:
			if item.type != 'item':
				continue
			my_data = await self.Libraries[0].read(item.name)
			if remove_path:
				assert item.name.startswith(path)
				tar_name = item.name[len(path):]
			else:
				tar_name = item.name
			info = tarfile.TarInfo(tar_name)
			my_data.seek(0, io.SEEK_END)
			info.size = my_data.tell()
			my_data.seek(0, io.SEEK_SET)
			info.mtime = time.time()
			tarobj.addfile(tarinfo=info, fileobj=my_data)

		tarobj.close()
		fileobj.seek(0)
		return fileobj


	async def subscribe(
		self,
		paths: typing.Union[str, typing.List[str]],
		target: typing.Union[str, tuple, None] = None,
	) -> None:
		"""
		Subscribe to changes for specified paths of the library.

		In order to notify on changes in the Library, this method must be used after the Library is ready.

		Args:
			paths (str | list[str]): Either single path or list of paths to be subscribed. All the paths must be absolute (start with '/').
			target: In which target to watch the changes. Possible values:
				- "global" to watch global path changes
				- "tenant" to watch path changes in tenants
				- ("tenant", TENANT_ID) to watch path changes in one specified tenant TENANT_ID

		Examples:
		```python
		class MyApplication(asab.Application):

			async def initialize(self):
				self.PubSub.subscribe("Library.ready!", self.on_library_ready
				self.PubSub.subscribe("Library.change!", self.on_library_change)

			async def on_library_ready(self, event_name, library=None):
				await self.LibraryService.subscribe(["/path1/","/path2/"])

			def on_library_change(self, message, provider, path):
				print("New changes in the library found by provider: '{}'".format(provider))
		```
		"""
		self._ensure_ready()

		if isinstance(paths, str):
			paths = [paths]
		for path in paths:
			if not path.startswith("/"):
				raise LibraryInvalidPathError(
					message="Directory path must start with '/' when subscribing to Library changes.",
					path=path,
				)

			for provider in self.Libraries:
				await provider.subscribe(path, target)

__init__(app, service_name, paths=None)

Initialize the LibraryService.

The library service is designed to "exist" in multiple instances, with different paths setups. For that reason, you have to provide unique service_name and there is no default value for that.

If paths are not provided, they are fetched from [library]providers configuration.

Parameters:

Name	Type	Description	Default
app	Application	The ASAB Application.	required
service_name	str	A unique name of the service.	required
paths	str | list[str] | None	Either single path or list of paths with which LibraryService is connected.	None

Source code in asab/library/service.py

def __init__(
	self,
	app: Application,
	service_name: str,
	paths: typing.Union[str, typing.List[str], None] = None
):
	"""
	Initialize the LibraryService.

	The library service is designed to "exist" in multiple instances,
	with different `paths` setups.
	For that reason, you have to provide unique `service_name`
	and there is no _default_ value for that.

	If `paths` are not provided, they are fetched from `[library]providers` configuration.

	Args:
		app: The ASAB Application.
		service_name: A unique name of the service.
		paths (str | list[str] | None ): Either single path or list of paths with which LibraryService is connected.
	"""

	super().__init__(app, service_name)
	self.Libraries: list[LibraryProviderABC] = []
	self.Disabled: dict = {}
	self.DisabledPaths: list = []
	self.Favorites: dict = {}
	self.FavoritePaths: list = []


	if paths is None:
		# load them from configuration
		try:
			paths = Config.getmultiline("library", "providers")
		except configparser.NoOptionError:
			L.critical("'providers' option is not present in configuration section 'library'.")
			raise SystemExit("Exit due to a critical configuration error.")

	# paths can be string if specified as argument
	if isinstance(paths, str):
		paths = re.split(r"\s+", paths)

	for layer, path in enumerate(paths):
		# Create library for each layer of paths
		self._create_library(path, layer)

	app.PubSub.subscribe("Application.tick/60!", self._on_tick60)

check_disabled(path)

Check if the item specified in path is disabled, either globally or for the specified tenant.

WARNING: When checking for items disabled for a tenant, it must be set in context variable before using this function! If it is not set automatically (e.g. from web request), it must be set manually.

Example:

1. Is path disabled for a specific tenant?

   ```python
   try:
           tenant_ctx = asab.contextvars.Tenant.set(tenant)
           disabled = self.LibraryService.check_disabled(path)
           ...
   finally:
           asab.contextvars.Tenant.reset(tenant_ctx)
   ```
   

2. Is path disabled globally?

   ```python
   disabled = self.LibraryService.check_disabled(path)
   ```
   

Parameters:

Name	Type	Description	Default
path	str	Path to the item to be checked.	required

Returns:

Type	Description
bool	True if the item is disabled for the tenant.

Source code in asab/library/service.py

def check_disabled(self, path: str) -> bool:
	"""
	Check if the item specified in path is disabled, either globally or for the specified tenant.

	**WARNING:** When checking for items disabled for a tenant,
	it must be set in context variable before using this function!
	If it is not set automatically (e.g. from web request), it must be set manually.

	Example:

	1. Is path disabled for a specific tenant?

		```python
		try:
			tenant_ctx = asab.contextvars.Tenant.set(tenant)
			disabled = self.LibraryService.check_disabled(path)
			...
		finally:
			asab.contextvars.Tenant.reset(tenant_ctx)
		```

	2. Is path disabled globally?

		```python
		disabled = self.LibraryService.check_disabled(path)
		```

	Args:
		path (str): Path to the item to be checked.

	Returns:
		`True` if the item is disabled for the tenant.
	"""
	if not isinstance(path, str) or not path:
		raise LibraryInvalidPathError(
			message="Argument 'path' must be a non-empty string.",
			path=path,
		)
	try:
		tenant = Tenant.get()
	except LookupError:
		tenant = None

	# First check disabled by path
	for dp, disabled in self.DisabledPaths:
		if path.startswith(dp):
			if '*' in disabled:
				# Path is disabled for everybody
				return True

			if tenant is not None and tenant in disabled:
				# Path is disabled for a specified tenant
				return True

	# Then check for a specific item entries

	disabled = self.Disabled.get(path)

	if disabled is None:
		return False

	if '*' in disabled:
		# Item is disabled for everybody
		return True

	if tenant is not None and tenant in disabled:
		# Item is disabled for a specified tenant
		return True

	return False

check_favorite(path, inherit=False)

Check if path is marked as favorite for the current tenant (or globally via '*').

If inherit is True and a parent folder is favorited for the tenant/'*', any child path under that folder is considered favorited.

WARNING: Tenant must be set in the context variable before using this function if you want tenant-aware behavior (same as check_disabled).

Examples:

1) Tenant-aware check:

    ```python
    try:
            tenant_ctx = asab.contextvars.Tenant.set(tenant)
            is_fav = self.LibraryService.check_favorite(path)
            ...
    finally:
            asab.contextvars.Tenant.reset(tenant_ctx)
    ```

2) Global-only check (any tenant):

    ```python
    is_fav = self.LibraryService.check_favorite(path)
    ```

Parameters:

Name	Type	Description	Default
path	str	Path to the item or folder to be checked.	required
inherit	bool	If True, a favorited folder marks all descendants as favorited.	False

Returns:

Name	Type	Description
bool	bool	True if favorited for current tenant or globally.

Source code in asab/library/service.py

def check_favorite(self, path: str, inherit: bool = False) -> bool:
	"""
	Check if `path` is marked as favorite for the current tenant (or globally via '*').

	If `inherit` is True and a parent *folder* is favorited for the tenant/'*',
	any child path under that folder is considered favorited.

	**WARNING:** Tenant must be set in the context variable before using this function
	if you want tenant-aware behavior (same as check_disabled).

	Examples:

	1) Tenant-aware check:

		```python
		try:
			tenant_ctx = asab.contextvars.Tenant.set(tenant)
			is_fav = self.LibraryService.check_favorite(path)
			...
		finally:
			asab.contextvars.Tenant.reset(tenant_ctx)
		```

	2) Global-only check (any tenant):

		```python
		is_fav = self.LibraryService.check_favorite(path)
		```

	Args:
		path (str): Path to the item or folder to be checked.
		inherit (bool): If True, a favorited folder marks all descendants as favorited.

	Returns:
		bool: True if favorited for current tenant or globally.
	"""
	if not isinstance(path, str) or not path:
		raise LibraryInvalidPathError(
			message="Argument 'path' must be a non-empty string.",
			path=path,
		)

	try:
		tenant = Tenant.get()
	except LookupError:
		tenant = None

	# 1) Folder favorites
	#    - Exact match if not inheriting
	#    - Prefix match if inheriting (folder favorite applies to children)
	if inherit:
		for fp, fav_tenants in self.FavoritePaths:
			if path.startswith(fp):
				if '*' in fav_tenants:
					return True
				if tenant is not None and tenant in fav_tenants:
					return True
	else:
		# exact folder favorite (no inheritance)
		for fp, fav_tenants in self.FavoritePaths:
			if path == fp:
				if '*' in fav_tenants:
					return True
				if tenant is not None and tenant in fav_tenants:
					return True

	# 2) Exact item favorites
	fav_tenants = self.Favorites.get(path)
	if fav_tenants is None:
		return False

	if '*' in fav_tenants:
		return True

	if tenant is not None and tenant in fav_tenants:
		return True

	return False

export(path='/', remove_path=False) async

Return a file-like stream containing a gzipped tar archive of the library contents of the path.

Parameters:

Name	Type	Description	Default
path	str	The path to export.	'/'
tenant	str | None	The tenant to use for the operation.	required
remove_path	bool	If True, the path will be removed from the tar file.	False

Returns:

Type	Description
IO	A file object containing a gzipped tar archive.

Source code in asab/library/service.py

async def export(self, path: str = "/", remove_path: bool = False) -> typing.IO:
	"""
	Return a file-like stream containing a gzipped tar archive of the library contents of the path.

	Args:
		path: The path to export.
		tenant (str | None ): The tenant to use for the operation.
		remove_path: If `True`, the path will be removed from the tar file.

	Returns:
		A file object containing a gzipped tar archive.
	"""
	self._ensure_ready()

	_validate_path_directory(path)

	fileobj = tempfile.TemporaryFile()
	tarobj = tarfile.open(name=None, mode='w:gz', fileobj=fileobj)

	items = await self._list(path, providers=self.Libraries[:1])
	recitems = list(items[:])

	while len(recitems) > 0:

		item = recitems.pop(0)
		if item.type != 'dir':
			continue

		child_items = await self._list(item.name, providers=item.providers)
		items.extend(child_items)
		recitems.extend(child_items)

	for item in items:
		if item.type != 'item':
			continue
		my_data = await self.Libraries[0].read(item.name)
		if remove_path:
			assert item.name.startswith(path)
			tar_name = item.name[len(path):]
		else:
			tar_name = item.name
		info = tarfile.TarInfo(tar_name)
		my_data.seek(0, io.SEEK_END)
		info.size = my_data.tell()
		my_data.seek(0, io.SEEK_SET)
		info.mtime = time.time()
		tarobj.addfile(tarinfo=info, fileobj=my_data)

	tarobj.close()
	fileobj.seek(0)
	return fileobj

find(path) async

Search for files with a specific name within a library, using the provided path.

The method traverses the library directories, looking for files that match the given filename. It returns a list of paths leading to these files, empty if no items are found.

Parameters:

Name	Type	Description	Default
path	str	Location of the file in Library. It must start with a forward slash and include the filename. Example: '/Dashboards/Cisco/Overview.json'	required

Returns:

Type	Description
List[str]	typing.List[str]: A list of paths to the found files. If no files are found, the list will be empty.

Source code in asab/library/service.py

async def find(self, path: str) -> typing.List[str]:
	"""
	Search for files with a specific name within a library, using the provided path.

	The method traverses the library directories, looking for files that match the given filename.
	It returns a list of paths leading to these files, empty if no items are found.

	Args:
		path (str): Location of the file in Library. It must start with a forward slash and include the filename. Example: '/Dashboards/Cisco/Overview.json'

	Returns:
		typing.List[str]: A list of paths to the found files. If no files are found, the list will be empty.
	"""
	_validate_path_item(path)

	results = []
	for library in self.Libraries:
		found_files = await library.find(path)
		if found_files:
			results.extend(found_files)
	return results

get_item_metadata(path) async

Retrieve metadata for a specific file in the library, including its target.

Parameters:

Name	Type	Description	Default
path	str	The absolute path of the file to retrieve metadata for. Must start with '/' and include a filename with an extension.	required

Returns:

Name	Type	Description
dict	Optional[dict]	Metadata for the specified file, including target, or None if not found.

Source code in asab/library/service.py

async def get_item_metadata(self, path: str) -> typing.Optional[dict]:
	"""
	Retrieve metadata for a specific file in the library, including its `target`.

	Args:
		path (str): The absolute path of the file to retrieve metadata for.
					Must start with '/' and include a filename with an extension.

	Returns:
		dict: Metadata for the specified file, including `target`, or None if not found.
	"""

	# Validate the path format
	_validate_path_item(path)

	# Split into directory and filename
	directory, filename = os.path.split(path)

	if not directory or not filename:
		L.warning("Invalid path '{}': missing directory or filename.".format(path))
		return None
	# Ensure directory ends with '/'
	if not directory.endswith('/'):
		directory += '/'

	try:
		# Fetch all items in the directory
		items = await self.list(directory)
	except Exception as e:
		L.warning("Failed to list items in directory '{}': {}".format(directory, e))
		return None

	# Use dictionary for faster lookup
	items_dict = {item.name: item for item in items}

	# Retrieve the item by path
	item = items_dict.get(path)
	if item and item.type == "item":
		# Match found; return metadata including `target`
		return {
			"name": item.name,
			"type": item.type,
			"layers": item.layers,
			"providers": item.providers,
			"disabled": item.disabled,
			"favorite": item.favorite,
			"override": item.override,
		}

	# Item not found
	L.info("Item '{}' not found in directory '{}'.".format(filename, directory))
	return None

is_ready()

Check if all the library providers are ready.

Returns:

Type	Description
bool	True if every provider is ready; if even one provider is not, returns False.

Source code in asab/library/service.py

def is_ready(self) -> bool:
	"""
	Check if all the library providers are ready.

	Returns:
		True if every provider is ready; if even one provider is not, returns False.
	"""
	if not self.Libraries:
		return False

	for provider in self.Libraries:
		if not provider.IsReady:
			return False
	return True

list(path='/', recursive=False) async

List the directory of the library specified by the path that are enabled for the specified tenant. This method can be used only after the Library is ready.

WARNING: Tenant must be set in the context variable! If it is not set automatically (e.g. from web request), it must be set manually.

Example:

try:
        tenant_ctx = asab.contextvars.Tenant.set(tenant)
        items = self.LibraryService.list(path)
        ...
finally:
        asab.contextvars.Tenant.reset(tenant_ctx)

Parameters:

Name	Type	Description	Default
path	str	Path to the directory.	'/'
recursive	bool	If True, return a list of items located at path and its subdirectories.	False

Returns:

Type	Description
List[LibraryItem]	List of items that are enabled for the tenant.

Source code in asab/library/service.py

async def list(self, path: str = "/", recursive: bool = False) -> typing.List[LibraryItem]:
	"""
	List the directory of the library specified by the path that are enabled for the specified tenant.
	This method can be used only after the Library is ready.

	**WARNING:** Tenant must be set in the context variable!
	If it is not set automatically (e.g. from web request), it must be set manually.

	Example:

	```python
	try:
		tenant_ctx = asab.contextvars.Tenant.set(tenant)
		items = self.LibraryService.list(path)
		...
	finally:
		asab.contextvars.Tenant.reset(tenant_ctx)
	```

	Args:
		path (str): Path to the directory.
		recursive (bool): If `True`, return a list of items located at `path` and its subdirectories.

	Returns:
		List of items that are enabled for the tenant.
	"""
	self._ensure_ready()

	_validate_path_directory(path)

	# List requested level using all available providers
	items = await self._list(path, providers=self.Libraries)

	if recursive:
		# If recursive scan is requested, then iterate thru list of items
		# find 'dir' types there and list them.
		# Output of this list is attached to the list for recursive scan
		# and also to the final output
		recitems = list(items[:])

		while len(recitems) > 0:

			item = recitems.pop(0)
			if item.type != 'dir':
				continue

			child_items = await self._list(item.name, providers=item.providers)
			items.extend(child_items)
			recitems.extend(child_items)

	return items

open(path) async

Read the content of the library item specified by path in a SAFE way, protected by a context manager/with statement. This method can be used only after the Library is ready.

Example:

async with self.LibraryService.open(path) as io:
        if io is None:
                return None
        text = b.read().decode("utf-8")

Source code in asab/library/service.py

@contextlib.asynccontextmanager
async def open(self, path: str):
	"""
	Read the content of the library item specified by `path` in a SAFE way, protected by a context manager/with statement.
	This method can be used only after the Library is ready.

	Example:

	```python
	async with self.LibraryService.open(path) as io:
		if io is None:
			return None
		text = b.read().decode("utf-8")
	```
	"""
	self._ensure_ready()

	_validate_path_item(path)

	# Same functionality as in read() method
	itemio = None
	disabled = self.check_disabled(path)
	if not disabled:
		for library in self.Libraries:
			itemio = await library.read(path)
			if itemio is not None:
				break

	if itemio is None:
		yield itemio
	else:
		try:
			yield itemio
		finally:
			itemio.close()

read(path) async

THIS IS OBSOLETED METHOD, USE open(...) !!!

Read the content of the library item specified by path. This method can be used only after the Library is ready.

Parameters:

Name	Type	Description	Default
path	str	Path to the file, LibraryItem.name can be used directly.	required
tenant	str | None	The tenant to apply. If not specified, the global access is assumed.	required

Returns:

Type	Description
IO | None	Readable stream with the content of the library item. None is returned if the item is not found or if it is disabled (either globally or for the specified tenant).

Example:

itemio = await library.read('/path', 'tenant')
if itemio is not None:
        with itemio:
                return itemio.read()

Source code in asab/library/service.py

async def read(self, path: str) -> typing.Optional[typing.IO]:
	"""
	THIS IS OBSOLETED METHOD, USE `open(...)` !!!

	Read the content of the library item specified by `path`. This method can be used only after the Library is ready.

	Args:
		path (str): Path to the file, `LibraryItem.name` can be used directly.
		tenant (str | None): The tenant to apply. If not specified, the global access is assumed.

	Returns:
		( IO | None ): Readable stream with the content of the library item. `None` is returned if the item is not found or if it is disabled (either globally or for the specified tenant).

	Example:

	```python
	itemio = await library.read('/path', 'tenant')
	if itemio is not None:
		with itemio:
			return itemio.read()
	```
	"""
	self._ensure_ready()

	LogObsolete.warning("Method 'LibraryService.read()' is obsolete. Use 'LibraryService.open()' method instead.")
	_validate_path_item(path)

	if self.check_disabled(path):
		return None

	for library in self.Libraries:
		itemio = await library.read(path)
		if itemio is None:
			continue
		return itemio

	return None

subscribe(paths, target=None) async

Subscribe to changes for specified paths of the library.

In order to notify on changes in the Library, this method must be used after the Library is ready.

Parameters:

Name	Type	Description	Default
paths	str | list[str]	Either single path or list of paths to be subscribed. All the paths must be absolute (start with '/').	required
target	Union[str, tuple, None]	In which target to watch the changes. Possible values: - "global" to watch global path changes - "tenant" to watch path changes in tenants - ("tenant", TENANT_ID) to watch path changes in one specified tenant TENANT_ID	None

Examples:

class MyApplication(asab.Application):

        async def initialize(self):
                self.PubSub.subscribe("Library.ready!", self.on_library_ready
                self.PubSub.subscribe("Library.change!", self.on_library_change)

        async def on_library_ready(self, event_name, library=None):
                await self.LibraryService.subscribe(["/path1/","/path2/"])

        def on_library_change(self, message, provider, path):
                print("New changes in the library found by provider: '{}'".format(provider))

Source code in asab/library/service.py

async def subscribe(
	self,
	paths: typing.Union[str, typing.List[str]],
	target: typing.Union[str, tuple, None] = None,
) -> None:
	"""
	Subscribe to changes for specified paths of the library.

	In order to notify on changes in the Library, this method must be used after the Library is ready.

	Args:
		paths (str | list[str]): Either single path or list of paths to be subscribed. All the paths must be absolute (start with '/').
		target: In which target to watch the changes. Possible values:
			- "global" to watch global path changes
			- "tenant" to watch path changes in tenants
			- ("tenant", TENANT_ID) to watch path changes in one specified tenant TENANT_ID

	Examples:
	```python
	class MyApplication(asab.Application):

		async def initialize(self):
			self.PubSub.subscribe("Library.ready!", self.on_library_ready
			self.PubSub.subscribe("Library.change!", self.on_library_change)

		async def on_library_ready(self, event_name, library=None):
			await self.LibraryService.subscribe(["/path1/","/path2/"])

		def on_library_change(self, message, provider, path):
			print("New changes in the library found by provider: '{}'".format(provider))
	```
	"""
	self._ensure_ready()

	if isinstance(paths, str):
		paths = [paths]
	for path in paths:
		if not path.startswith("/"):
			raise LibraryInvalidPathError(
				message="Directory path must start with '/' when subscribing to Library changes.",
				path=path,
			)

		for provider in self.Libraries:
			await provider.subscribe(path, target)

asab.library.item.LibraryItem dataclass

The data class that contains the info about a specific item in the library.

Attributes:

Name	Type	Description
name	str	The absolute path of the Item. It can be directly fed into LibraryService.read(...).
type	str	Can be either dir if the Item is a directory or item if Item is of any other type.
layer	int	The number of highest layer in which this Item is found. The higher the number, the lower the layer is.
providers	list	List of LibraryProvider objects containing this Item.
disabled	bool	True if the Item is disabled, False otherwise. If the Item is disabled, LibraryService.read(...) will return None.
favorite	bool	True if the Item is marked as a favorite.
override	int	If True, this item is marked as an override for the providers with the same Item name.
target	str	Specifies the target context, e.g., "tenant" or "global". Defaults to "global".

Source code in asab/library/item.py

@dataclasses.dataclass
class LibraryItem:
    """
    The data class that contains the info about a specific item in the library.

    Attributes:
        name (str): The absolute path of the Item. It can be directly fed into `LibraryService.read(...)`.
        type (str): Can be either `dir` if the Item is a directory or `item` if Item is of any other type.
        layer (int): The number of highest layer in which this Item is found. The higher the number, the lower the layer is.
        providers (list): List of `LibraryProvider` objects containing this Item.
        disabled (bool): `True` if the Item is disabled, `False` otherwise. If the Item is disabled, `LibraryService.read(...)` will return `None`.
        favorite (bool): True if the Item is marked as a favorite.
        override (int): If `True`, this item is marked as an override for the providers with the same Item name.
        target (str): Specifies the target context, e.g., "tenant" or "global". Defaults to "global".
    """
    name: str
    type: str
    layers: typing.List[int]  # Now stores multiple layers in ascending order
    providers: list
    disabled: bool = False
    favorite: bool = False
    override: int = 0
    size: typing.Optional[int] = None  # Size is None by default, absent for directories