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.
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.
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.
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.
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.
Initializes the Library Service. Remember to specify a unique service_name.
When the Library is initialized, Library.ready! PubSub message is emitted.
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.
list() method returns list of LibraryItems. For more information, see the reference section.
item.type can be either 'item' or 'dir'.
read() coroutine returns item IO object or None if the file is disabled.
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.
Some providers are able to detect changes of the library items.
Example
classMyApplication(asab.Application):asyncdefinitialize(self):self.PubSub.subscribe("Library.ready!",self.on_library_readyself.PubSub.subscribe("Library.change!",self.on_library_change)asyncdefon_library_ready(self,event_name,library=None):awaitself.LibraryService.subscribe(["/asab"])#(1)!defon_library_change(self,message,provider,path):#(2)!print("New changes in the library found by provider: '{}'".format(provider))
self.LibraryService.subscribe() method takes either a single path as a string or multiple paths in list and watches for changes in them.
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.
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.
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:
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:
Go to Settings > Repository > Deploy tokens section in your
repository. (Note that you have to possess a "Maintainer" or
"Owner" role for the repository.)
Expand the "Deploy tokens" section. The list of current Active
Deploy Tokens will be displayed.
Complete the fields and scopes. We recommend a custom "username",
as you will need it later for the URL in the configuration.
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:
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:
The libsreg provider downloads the content from the distribution URL.
The distribution URL points to HTTP(S) server where content archives are published.
*.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.
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!.
classLibraryService(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=[]ifpathsisNone:# load them from configurationtry:paths=Config.getmultiline("library","providers")exceptconfigparser.NoOptionError:L.critical("'providers' option is not present in configuration section 'library'.")raiseSystemExit("Exit due to a critical configuration error.")# paths can be string if specified as argumentifisinstance(paths,str):paths=re.split(r"\s+",paths)forlayer,pathinenumerate(paths):# Create library for each layer of pathsself._create_library(path,layer)app.PubSub.subscribe("Application.tick/60!",self._on_tick60)asyncdeffinalize(self,app):whilelen(self.Libraries)>0:lib=self.Libraries.pop(-1)awaitlib.finalize(self.App)asyncdef_on_tick60(self,message_type):awaitself._read_disabled()def_create_library(self,path,layer):library_provider=Noneifpath.startswith('zk://')orpath.startswith('zookeeper://'):from.providers.zookeeperimportZooKeeperLibraryProviderlibrary_provider=ZooKeeperLibraryProvider(self,path,layer)elifpath.startswith('./')orpath.startswith('/')orpath.startswith('file://'):from.providers.filesystemimportFileSystemLibraryProviderlibrary_provider=FileSystemLibraryProvider(self,path,layer)elifpath.startswith('azure+https://'):from.providers.azurestorageimportAzureStorageLibraryProviderlibrary_provider=AzureStorageLibraryProvider(self,path,layer)elifpath.startswith('git+'):from.providers.gitimportGitLibraryProviderlibrary_provider=GitLibraryProvider(self,path,layer)elifpath.startswith('libsreg+'):from.providers.libsregimportLibsRegLibraryProviderlibrary_provider=LibsRegLibraryProvider(self,path,layer)elifpath==''orpath.startswith("#")orpath.startswith(";"):# This is empty or commented linereturnelse:L.error("Incorrect/unknown provider for '{}'".format(path))raiseSystemExit("Exit due to a critical configuration error.")self.Libraries.append(library_provider)defis_ready(self)->bool:""" Check if all the libraries are ready. Returns: True if all libraries are ready, otherwise False. """iflen(self.Libraries)==0:returnFalsereturnfunctools.reduce(lambdax,provider:provider.IsReadyandx,self.Libraries,True)asyncdef_set_ready(self,provider):iflen(self.Libraries)==0:returnif(provider==self.Libraries[0])andprovider.IsReady:awaitself._read_disabled()ifself.is_ready():L.log(LOG_NOTICE,"is ready.",struct_data={'name':self.Name})self.App.PubSub.publish("Library.ready!",self)elifnotprovider.IsReady:L.log(LOG_NOTICE,"is NOT ready.",struct_data={'name':self.Name})self.App.PubSub.publish("Library.not_ready!",self)asyncdeffind(self,path:str)->typing.Optional[typing.List[str]]:""" Searches 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, or `None` if no such files are found. Args: path (str): The path specifying the filename and its location within the library. The path should start with a forward slash and include the filename. Example: '/library/Templates/.setup.yaml' Returns: typing.Optional[typing.List[str]]: A list containing the paths to the found files, `or an `empty list` if no files are found that match the filename. """# File path must start with '/'assertpath[:1]=='/',"Item path '{}' must start with a forward slash (/). For example: /library/Templates/.setup.yaml".format(path)# File path must end with the filenameassertlen(os.path.splitext(path)[1])>0,"Item path '{}' must end with an extension. For example: /library/Templates/item.json".format(path)results=[]forlibraryinself.Libraries:found_files=awaitlibrary.find(path)iffound_files:results.extend(found_files)returnresultsasyncdefread(self,path:str,tenant:typing.Optional[str]=None)->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() ``` """# item path must start with '/'assertpath[:1]=='/',"Item path must start with a forward slash (/). For example: /library/Templates/item.json"# Item path must end with the extensionassertlen(os.path.splitext(path)[1])>0,"Item path must end with an extension. For example: /library/Templates/item.json"ifself.check_disabled(path,tenant=tenant):returnNoneforlibraryinself.Libraries:itemio=awaitlibrary.read(path)ifitemioisNone:continuereturnitemioreturnNone@contextlib.asynccontextmanagerasyncdefopen(self,path:str,tenant:typing.Optional[str]=None):""" 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.App.LibraryService.open(path) as b: if b is None: return None text = b.read().decode("utf-8") ``` """itemio=awaitself.read(path,tenant)ifitemioisNone:yielditemioelse:try:yielditemiofinally:itemio.close()asyncdeflist(self,path:str="/",tenant:typing.Optional[str]=None,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. Args: path (str): Path to the directory. tenant (str | None): If specified, items that are enabled for the tenant are filtered. 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. """# Directory path must start with '/'assertpath[:1]=='/',"Directory path must start with a forward slash (/). For example: /library/Templates/"# Directory path must end with '/'assertpath[-1:]=='/',"Directory path must end with a forward slash (/). For example: /library/Templates/"# Directory path cannot contain '//'assert'//'notinpath,"Directory path cannot contain double slashes (//). Example format: /library/Templates/"# List requested level using all available providersitems=awaitself._list(path,tenant,providers=self.Libraries)ifrecursive:# 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 outputrecitems=list(items[:])whilelen(recitems)>0:item=recitems.pop(0)ifitem.type!='dir':continuechild_items=awaitself._list(item.name,tenant,providers=item.providers)items.extend(child_items)recitems.extend(child_items)returnitemsasyncdef_list(self,path,tenant,providers):# Execute the list query in all providers in-parallelresult=awaitasyncio.gather(*[library.list(path)forlibraryinproviders],return_exceptions=True)items=[]uniq=dict()forressinresult:ifisinstance(ress,KeyError):# The path doesn't exists in the providercontinueifisinstance(ress,Exception):L.exception("Error when listing items from provider",exc_info=ress)continueforiteminress:item.disabled=self.check_disabled(item.name,tenant=tenant)# If the item already exists, merge or override itpitem=uniq.get(item.name)ifpitemisnotNone:pitem=uniq[item.name]ifpitem.type=='dir'anditem.type=='dir':# Directories are joinedpitem.providers.extend(item.providers)elifpitem.type=='item':fori,providerinenumerate(providers):ifproviderinitem.providers:index=ibreakpitem.override=index# Other item types are skippedelse:uniq[item.name]=itemitems.append(item)items.sort(key=lambdax:x.name)returnitemsasyncdef_read_disabled(self):# `.disabled.yaml` is read from the first configured library# It is applied on all libraries in the configuration.disabled=awaitself.Libraries[0].read('/.disabled.yaml')ifdisabledisNone:self.Disabled={}self.DisabledPaths=[]returntry:disabled=yaml.safe_load(disabled)exceptException:self.Disabled={}self.DisabledPaths=[]L.exception("Failed to parse '/.disabled.yaml'")returnifdisabledisNone:self.Disabled={}self.DisabledPaths=[]returnifisinstance(disabled,set):# This is for a backward compatibility (Aug 2023)self.Disabled={key:'*'forkeyinself.Disabled}self.DisabledPaths=[]returnself.Disabled={}self.DisabledPaths=[]fork,vindisabled.items():ifk.endswith('/'):self.DisabledPaths.append((k,v))else:self.Disabled[k]=v# Sort self.DisabledPaths from the shortest to longestself.DisabledPaths.sort(key=lambdax:len(x[0]))defcheck_disabled(self,path:str,tenant:typing.Optional[str]=None)->bool:""" Check if the item specified in path is disabled, either globally or for the specified tenant. Args: path (str): Path to the item to be checked. tenant (str | None): The tenant to apply. If not specified, the global access is assumed. Returns: `True` if the item is disabled for the tenant. """ifnotisinstance(path,str)ornotpath:raiseValueError("The 'path' must be a non-empty string.")# First check disabled by pathfordp,disabledinself.DisabledPaths:ifpath.startswith(dp):if'*'indisabled:# Path is disabled for everybodyreturnTrueiftenantisnotNoneandtenantindisabled:# Path is disabled for a specified tenantreturnTrue# Then check for a specific item entriesdisabled=self.Disabled.get(path)ifdisabledisNone:returnFalseif'*'indisabled:# Item is disabled for everybodyreturnTrueiftenantisnotNoneandtenantindisabled:# Item is disabled for a specified tenantreturnTruereturnFalseasyncdefexport(self,path:str="/",tenant:typing.Optional[str]=None,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. """# Directory path must start with '/'assertpath[:1]=='/',"Directory path must start with a forward slash (/). For example: /library/Templates/"# Directory path must end with '/'assertpath[-1:]=='/',"Directory path must end with a forward slash (/). For example: /library/Templates/"# Directory path cannot contain '//'assert'//'notinpath,"Directory path cannot contain double slashes (//). Example format: /library/Templates/"fileobj=tempfile.TemporaryFile()tarobj=tarfile.open(name=None,mode='w:gz',fileobj=fileobj)items=awaitself._list(path,tenant,providers=self.Libraries[:1])recitems=list(items[:])whilelen(recitems)>0:item=recitems.pop(0)ifitem.type!='dir':continuechild_items=awaitself._list(item.name,tenant,providers=item.providers)items.extend(child_items)recitems.extend(child_items)foriteminitems:ifitem.type!='item':continuemy_data=awaitself.Libraries[0].read(item.name)ifremove_path:assertitem.name.startswith(path)tar_name=item.name[len(path):]else:tar_name=item.nameinfo=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)returnfileobjasyncdefsubscribe(self,paths:typing.Union[str,typing.List[str]])->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 '/'). 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(["/alpha","/beta"]) def on_library_change(self, message, provider, path): print("New changes in the library found by provider: '{}'".format(provider)) ``` """ifisinstance(paths,str):paths=[paths]forpathinpaths:assertpath[:1]=='/',"Absolute path must be used when subscribing to the library changes"forproviderinself.Libraries:awaitprovider.subscribe(path)
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.
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=[]ifpathsisNone:# load them from configurationtry:paths=Config.getmultiline("library","providers")exceptconfigparser.NoOptionError:L.critical("'providers' option is not present in configuration section 'library'.")raiseSystemExit("Exit due to a critical configuration error.")# paths can be string if specified as argumentifisinstance(paths,str):paths=re.split(r"\s+",paths)forlayer,pathinenumerate(paths):# Create library for each layer of pathsself._create_library(path,layer)app.PubSub.subscribe("Application.tick/60!",self._on_tick60)
defcheck_disabled(self,path:str,tenant:typing.Optional[str]=None)->bool:""" Check if the item specified in path is disabled, either globally or for the specified tenant. Args: path (str): Path to the item to be checked. tenant (str | None): The tenant to apply. If not specified, the global access is assumed. Returns: `True` if the item is disabled for the tenant. """ifnotisinstance(path,str)ornotpath:raiseValueError("The 'path' must be a non-empty string.")# First check disabled by pathfordp,disabledinself.DisabledPaths:ifpath.startswith(dp):if'*'indisabled:# Path is disabled for everybodyreturnTrueiftenantisnotNoneandtenantindisabled:# Path is disabled for a specified tenantreturnTrue# Then check for a specific item entriesdisabled=self.Disabled.get(path)ifdisabledisNone:returnFalseif'*'indisabled:# Item is disabled for everybodyreturnTrueiftenantisnotNoneandtenantindisabled:# Item is disabled for a specified tenantreturnTruereturnFalse
asyncdefexport(self,path:str="/",tenant:typing.Optional[str]=None,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. """# Directory path must start with '/'assertpath[:1]=='/',"Directory path must start with a forward slash (/). For example: /library/Templates/"# Directory path must end with '/'assertpath[-1:]=='/',"Directory path must end with a forward slash (/). For example: /library/Templates/"# Directory path cannot contain '//'assert'//'notinpath,"Directory path cannot contain double slashes (//). Example format: /library/Templates/"fileobj=tempfile.TemporaryFile()tarobj=tarfile.open(name=None,mode='w:gz',fileobj=fileobj)items=awaitself._list(path,tenant,providers=self.Libraries[:1])recitems=list(items[:])whilelen(recitems)>0:item=recitems.pop(0)ifitem.type!='dir':continuechild_items=awaitself._list(item.name,tenant,providers=item.providers)items.extend(child_items)recitems.extend(child_items)foriteminitems:ifitem.type!='item':continuemy_data=awaitself.Libraries[0].read(item.name)ifremove_path:assertitem.name.startswith(path)tar_name=item.name[len(path):]else:tar_name=item.nameinfo=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)returnfileobj
Searches 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, or None if no such files are found.
Parameters:
Name
Type
Description
Default
path
str
The path specifying the filename and its location within the library.
The path should start with a forward slash and include the filename.
Example: '/library/Templates/.setup.yaml'
required
Returns:
Type
Description
Optional[List[str]]
typing.Optional[typing.List[str]]: A list containing the paths to the found files,
or anempty list` if no files are found that match the filename.
asyncdeffind(self,path:str)->typing.Optional[typing.List[str]]:""" Searches 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, or `None` if no such files are found. Args: path (str): The path specifying the filename and its location within the library. The path should start with a forward slash and include the filename. Example: '/library/Templates/.setup.yaml' Returns: typing.Optional[typing.List[str]]: A list containing the paths to the found files, `or an `empty list` if no files are found that match the filename. """# File path must start with '/'assertpath[:1]=='/',"Item path '{}' must start with a forward slash (/). For example: /library/Templates/.setup.yaml".format(path)# File path must end with the filenameassertlen(os.path.splitext(path)[1])>0,"Item path '{}' must end with an extension. For example: /library/Templates/item.json".format(path)results=[]forlibraryinself.Libraries:found_files=awaitlibrary.find(path)iffound_files:results.extend(found_files)returnresults
defis_ready(self)->bool:""" Check if all the libraries are ready. Returns: True if all libraries are ready, otherwise False. """iflen(self.Libraries)==0:returnFalsereturnfunctools.reduce(lambdax,provider:provider.IsReadyandx,self.Libraries,True)
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.
Parameters:
Name
Type
Description
Default
path
str
Path to the directory.
'/'
tenant
str | None
If specified, items that are enabled for the tenant are filtered.
None
recursive
bool
If True, return a list of items located at path and its subdirectories.
asyncdeflist(self,path:str="/",tenant:typing.Optional[str]=None,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. Args: path (str): Path to the directory. tenant (str | None): If specified, items that are enabled for the tenant are filtered. 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. """# Directory path must start with '/'assertpath[:1]=='/',"Directory path must start with a forward slash (/). For example: /library/Templates/"# Directory path must end with '/'assertpath[-1:]=='/',"Directory path must end with a forward slash (/). For example: /library/Templates/"# Directory path cannot contain '//'assert'//'notinpath,"Directory path cannot contain double slashes (//). Example format: /library/Templates/"# List requested level using all available providersitems=awaitself._list(path,tenant,providers=self.Libraries)ifrecursive:# 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 outputrecitems=list(items[:])whilelen(recitems)>0:item=recitems.pop(0)ifitem.type!='dir':continuechild_items=awaitself._list(item.name,tenant,providers=item.providers)items.extend(child_items)recitems.extend(child_items)returnitems
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.
@contextlib.asynccontextmanagerasyncdefopen(self,path:str,tenant:typing.Optional[str]=None):""" 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.App.LibraryService.open(path) as b: if b is None: return None text = b.read().decode("utf-8") ``` """itemio=awaitself.read(path,tenant)ifitemioisNone:yielditemioelse:try:yielditemiofinally:itemio.close()
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.
None
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).
asyncdefread(self,path:str,tenant:typing.Optional[str]=None)->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() ``` """# item path must start with '/'assertpath[:1]=='/',"Item path must start with a forward slash (/). For example: /library/Templates/item.json"# Item path must end with the extensionassertlen(os.path.splitext(path)[1])>0,"Item path must end with an extension. For example: /library/Templates/item.json"ifself.check_disabled(path,tenant=tenant):returnNoneforlibraryinself.Libraries:itemio=awaitlibrary.read(path)ifitemioisNone:continuereturnitemioreturnNone
Either single path or list of paths to be subscribed. All the paths must be absolute (start with '/').
required
Examples:
classMyApplication(asab.Application):asyncdefinitialize(self):self.PubSub.subscribe("Library.ready!",self.on_library_readyself.PubSub.subscribe("Library.change!",self.on_library_change)asyncdefon_library_ready(self,event_name,library=None):awaitself.LibraryService.subscribe(["/alpha","/beta"])defon_library_change(self,message,provider,path):print("New changes in the library found by provider: '{}'".format(provider))
asyncdefsubscribe(self,paths:typing.Union[str,typing.List[str]])->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 '/'). 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(["/alpha","/beta"]) def on_library_change(self, message, provider, path): print("New changes in the library found by provider: '{}'".format(provider)) ``` """ifisinstance(paths,str):paths=[paths]forpathinpaths:assertpath[:1]=='/',"Absolute path must be used when subscribing to the library changes"forproviderinself.Libraries:awaitprovider.subscribe(path)
@dataclasses.dataclassclassLibraryItem:""" 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`. override (int): If `True`, this item is marked as an override for the providers with the same Item name. """name:strtype:strlayer:intproviders:listdisabled:bool=Falseoverride:int=0# Default value for override is False
