sl.snippet_api.snippet package

class sl.snippet_api.snippet.AppLogAdapter(logger, extra=None)

Bases: LoggerAdapter

Custom logging for Dynamic Apps that includes extra formatting.

process(msg, kwargs)

Process the logging message and keyword arguments passed in to a logging call to insert contextual information. You can either manipulate the message itself, the keyword args or both. Return the message and kwargs modified (or not) to suit your needs.

Normally, you’ll only need to override this one method in a LoggerAdapter subclass for your specific needs.

Return type:: Tuple[str, dict]

sl.snippet_api.snippet.CacheAPI(dbc, cache_type, app_id, did, ttl, timestamp=None)

sl.snippet_api.snippet.CacheCipher: alias of CacheCipherV1

exception sl.snippet_api.snippet.CacheDataIsEmpty

Bases: CollectorCacheError

Raised when a cache entry is requested, but the entry does not exist.

exception sl.snippet_api.snippet.CacheDataIsExpired

Bases: CacheDataIsEmpty

Raised when a requested key is present but the expiration has passed.

This subclasses CacheDataIsEmpty because an expired entry is essentially data that no longer exists. The caller can catch either exception based on the use case.

exception sl.snippet_api.snippet.CacheKeyIsNone

Bases: CollectorCacheError

Raised when an empty cache key is used.

exception sl.snippet_api.snippet.CacheKeyNotFound(key)

Bases: CollectorCacheError, KeyError

Raised when a cache key is not found.

class sl.snippet_api.snippet.CacheManager(app=None, prefix='', dbc=None, ttl=None, expirations=True, in_memory=False, pickling=True, plaintext=True, json=False)

Bases: object

Class that can be used as a cache manager to handle the EM7 cache.

The defaults for this class will use db-caching unless otherwise specified.

__init__(app=None, prefix='', dbc=None, ttl=None, expirations=True, in_memory=False, pickling=True, plaintext=True, json=False)

Parameters:

app (Optional[DynamicApp]) – A snippet_application object. If passed, the dbc and ttl are reused from the app. This is here for backwards compatibility. Originally the base cache was structured this way, so the pattern was reused, then changed to have the dbc and ttl be parameterized.
ttl (Optional[int]) – The number of seconds from the time of writing for which the cache entry will be valid. The default is 3 times the app frequency.
dbc (Union[silo_cursor, SiloCursor, None]) – The database cursor. If none is provided, a local database cursor will be used.
expirations (Optional[bool]) – Use the expires field when reading the cache?
in_memory (Optional[bool]) – Use in-memory caching with database caching. When True, a global time-to-live based cache will be used for storing frequently used items. The items will be available to CacheManager objects across apps in the same process.
pickling (Optional[bool]) – Pickle the values? If not, store them as strings. This is the default if not specified in calls to read() and write().
plaintext (Optional[bool]) – Store the keys and values in plaintext? If no, encrypt them both.
json (Optional[bool]) – This is used to json serialize data. If json=True then we will serialize and deserialize objects in write and read functions, don’t work with pickling enabled.

Note

silo_common.maint.task.collector_dynapp_cache_prune will delete expired cache entries

Note

The actual cache object is created lazily. No database connection is established until the CacheManager.cache object is used.

property cache

delete(key_pattern, metadata=None, prefix=True)

Delete a cache entry based on the key

Parameters:

key_pattern (str) – key pattern to pass to LIKE SQL clause
metadata (dict) – Metadata containing a key to the cache value
prefix (bool) – Prepend the instances prefix as part of the cache key

get_options(pickling, plaintext, json)

Get the options to use for the Cache

Parameters:

pickling (bool) – Determine if the operation is pickle
plaintext (bool) – Determine if the operation is plaintext
json (bool) – Determine if the operation uses json data

Return type:

Tuple[bool, bool, bool]

read(cachekey, prefix=True, metadata=False, pickling=None, return_expiry=False, plaintext=None, json=None)

Read from Cache given a key

Parameters:

cachekey (str) – The key to use to retrieve the value from the cache.
prefix (Optional[bool]) – Prepend the instance’s prefix to the cache key? If False, use the key as-is.
metadata (Optional[bool]) – Look up the metadata cache entry that corresponds with the cache key
pickling (Optional[bool]) – Unpickle the retrieved value? If False, assume it is string.
return_expiry (Optional[bool]) – Return the expiration time with the result? If True, return type is a tuple: (results, expires).
plaintext (Optional[bool]) – Were the key and value stored in plaintext? If False, decrypt them both.
json (Optional[bool]) – This is used to read data in json serialized format. If json=True then we will deserialize by loads after reading from cache. Works after decoding data when is required.

Raises:

CacheKeyIsNone – Raised when no cache key is provided.
CacheKeyNotFound – Raised when cache key is not found.
CacheDataIsExpired – Raised when self.expirations==True and the expiration on the entry has passed.

Return type:

Union[str, Tuple[str, datetime]]

Returns:

Decoded string value. If return_expiry==True, also return the expiration timestamp.

read_by_prefix(prefix=None, pickling=None, return_expiry=False)

Given a prefix, return all cache entries that match.

Note that this method will not retrieve encrypted entries, and so does not attempt to decrypt anything. Since the keys are obfuscated, the prefix will never (well, almost never) match.

Parameters:

prefix (string) – prefix value for cache
pickling (bool) – unserialize value or not
return_expiry (bool) – Include the expiration in the result

Returns:

values List of (key, value) that met the criteria. (key, value, expires) if requested

Return type:

list

read_json(results, cachekey)

Deserialize the result from the cache entry

Parameters:

results – Results when looking up the cache key
cachekey – The key to use to retrieve the value from the cache.

Return type:

object

read_metadata(results, pickling)

Read metadata from the cache entry

Parameters:

results (Dict) – Results when looking up the cache key
pickling (bool) – Determine if the operation is pickle

Return type:

object

write(cachekey, cdata, metadata=None, prefix=True, pickling=None, ttl=None, plaintext=None, json=None)

Write a value to the cache.

Parameters:

cachekey (str) – The key to use to write the value to the cache.
cdata – Object or string. If object, set pickling=True
metadata (dict) – Metadata where the value’s cachekey can be found
prefix (bool) – Prepend the instances prefix as part of the cache key
pickling (bool) – Pickle the stored value? If not, assume it is string.
ttl (int) – The time to live for the new cache entry. If not provided, use the value for the instance.
plaintext (bool) – Store the key and value in plaintext? If not, encrypt them both.
json (bool) – This is used to give data json format. If json=True then we will serialize by dumps before caching. Works before encrypting data when is required.

Raises:

CacheKeyIsNone – Raised when no cache key is provided.

Returns:

The key used to store the value. If plaintext==False, the hashed key is returned.

exception sl.snippet_api.snippet.CachePrefixIsNone

Bases: CollectorCacheError

Raised when an empty cache key is used.

exception sl.snippet_api.snippet.CollectorCacheError

Bases: SLError

Base exception for cache errors.

exception sl.snippet_api.snippet.CredentialError: Bases: DeviceError

exception sl.snippet_api.snippet.CryptError

Bases: Exception

Base encryption and decryption error

exception sl.snippet_api.snippet.DeviceError: Bases: SLError

exception sl.snippet_api.snippet.EncryptError

Bases: CollectorCacheError

Raised when a data cannot be encrypted to save in cache.

exception sl.snippet_api.snippet.InvalidHeader

Bases: NotEncryptedError, TypeError

Raised when a non-string is passed for the crypt header

exception sl.snippet_api.snippet.InvalidObject

Bases: CryptError, TypeError

Raised when a non-string is passed for crypt operations

exception sl.snippet_api.snippet.InvalidOptions

Bases: CollectorCacheError

Raised when some configuration is not allowed.

class sl.snippet_api.snippet.LogManager(app_id=None, did=None, msg_id=None)

Bases: object

A context manager that configures the app logger.

This context manager configures a logger with a header specific to the app in which it is running, sets the header as the default header for subsequent callers inside libraries, then resets the default header on exit.

Usage example in a snippet:

logger = logging.getLogger(__name__)
logger.debug("does not have a header")

with LogManager(self.app_id, self.did, self.gmtime) as clogger:
    clogger.info("this message will have a header")
    lib.collect()

logger.info("does not have a header")

Libraries called from within the context manager can get a handle to a logger with the same header like this:

logger = app_logging.getLogger(__name__)
logger = app_logging.getCurrentAppLogger(logger)
logger.info("this message will have a header if inside an context manager")

__init__(app_id=None, did=None, msg_id=None)

Create the log manager.

Parameters:

app_id (Optional[str]) – A snippet app ID.
did (Optional[int]) – The device ID.
msg_id (Optional[str]) – A string that will identify messages from this logging instance.

exception sl.snippet_api.snippet.MalformedConfig: Bases: Exception

exception sl.snippet_api.snippet.MetadataKeyMissing

Bases: CollectorCacheError

Raised when attempting to read metadata without a metadata dict

exception sl.snippet_api.snippet.NotEncryptedError

Bases: CryptError

Raised when attempting to decrypt an unencrypted string.

This error means the string did not complete the process or was never encrypted at all. Some possible reasons for this error are: not base64 encoded, not encrypted, not convertible to a byte-string, etc.

exception sl.snippet_api.snippet.SLError: Bases: Exception

exception sl.snippet_api.snippet.SuppressedException

Bases: SLError

Exception that can be raised and will only be logged by error_manager.

sl.snippet_api.snippet.aes_decrypt(key, enc)

Decrypt a string with the key and header.

Hash and reduce the key, base64 decode the raw string, decrypt the raw string, and then remove any padding.

Parameters:

key (str) – a unicode string decryption key
enc (AnyStr) – an encrypted byte-string to be decrypted

Return type:

str

Returns:

Decrypted byte-string.

sl.snippet_api.snippet.aes_decrypt_header(key, enc, header)

Decrypt a string with the key and header.

Parameters:

key (str) – see aes_decrypt
enc (AnyStr) – see aes_decrypt
header (str) – an identifier to remove from the encoded string before decrypting

Raises:

InvalidHeader – Raised when the header is not a string, or the header is missing from the encrypted string.

Return type:

str

Returns:

The string, header removed from the raw string, and the raw string decrypted.

sl.snippet_api.snippet.aes_encrypt(key, raw)

Encrypt a string with the key and header.

Hash the key and reduce it, then create a random iv, pad and encrypt the raw string, then base64 encode it.

Parameters:

key (str) – a unicode string encryption key
raw (AnyStr) – a byte-string or decoded string to be encrypted

Return type:

bytes

Returns:

Base64 encoded and aes encrypted byte-string.

sl.snippet_api.snippet.aes_encrypt_header(key, raw, header)

Encrypt a string with the key and header.

Parameters:

key (str) – see aes_encrypt
raw (AnyStr) – see aes_encrypt
header (str) – an identifier to add to the encoded string after encryption

Return type:

bytes

Returns:

Encrypted byte-string, and header added to the raw final string.

sl.snippet_api.snippet.cred_type(cred_id, dbc=None)

Gets the credential type

Parameters:

cred_id (int) – the ID of the credential being aligned
dbc (Union[silo_cursor, SiloCursor, None]) – The database cursor. If none is provided, a local database cursor will be used

Return type:

Optional[str]

Returns:

Credential Type or None

sl.snippet_api.snippet.credential_type_name(cred_type, cred_id=None, skip_lookup=False)

Lookup the Credential Type for a given credential

Parameters:

cred_type (int) – Credential Type ID
cred_id (Optional[int]) – ID of the Credential to be used for universal credential lookups
skip_lookup (Optional[bool]) – Skip the universal credential lookup and return the friendly name

Return type:

str

Returns:

SL1 type name for the credential

sl.snippet_api.snippet.dbc_cursor(max_retry_time=5, legacy=False, **k)

Get a local database cursor.

Build parameters and call the function with those parameters. Based on the legacy parameter, decide to use a legacy cursor or a common cursor

Parameters:

max_retry_time (Optional[int]) – The maximum amount of time (in seconds) to attempt connecting.
legacy (Optional[bool]) – Whether the type of connection is legacy or not

Return type:

Union[silo_cursor, SiloCursor]

Returns:

If legacy cursor it will return a LegacySiloCursor, if not return SiloCursor

sl.snippet_api.snippet.deprecated(reason)

This is a decorator which can be used to mark functions as deprecated. It will result in a warning being emitted when the function is used.

Parameters:: reason (str) – Reason for the deprecation
Return type:: Callable
Returns:: A decorator which can be used to mark deprecated functions

sl.snippet_api.snippet.deprecation_warning(reason, stacklevel=2)

Sends a DeprecationWarning with the given reason

Enables DeprecationWarnings to always show, sends the deprecation warning, then restores default functionality.

Parameters:

reason (str) – Reason for the deprecation
stacklevel (Optional[int]) – Number of stacks to show with the deprecation message. Default: 2

Return type:

None

class sl.snippet_api.snippet.error_manager(app, mode=None, root_alerts_only=True, local_variables=False)

Bases: object

Class that can be used as a context manager that handles errors from snippets and routes them to the appropriate places. The platform should handle snippet exceptions, but sometimes we want to handle them from the app side.

Variables:

app – The dynamic app the context manager is protecting.
mode – The context/mode in which the error manager is used. Values are “discovery” and “dev”. “discovery” prevents the logging of any exceptions. “dev” logs the local variables of the origin traceback.
root_alerts_only – If True, then it will add a 3rd party device log to root device component when a DeviceError exception happened in root or a child device. If False, it will add an internal device log to the component device where DeviceError exception happened (previous behavior).

For an unhandled error:

Log to an error log that can handle the full stack trace.
Log to silo.log where the stack trace will get truncated, and point to the full message.
Create an alert on the device.

sl.snippet_api.snippet.fix_encoding(ob)

Converts the provided object or its component parts into ascii when given a unicode source. Dict keys and values, list and tuple values, and unicode strings are converted. All other object types are returned exactly as supplied

Parameters:: ob (object) – Object to encode to ascii
Return type:: Any
Returns:: Ascii encoded object

sl.snippet_api.snippet.getAppHeader(app_id, did, msg_id)

Create a header for a AppLogAdapter

Parameters:

app_id (str) – A snippet app id
did (int) – The device id
msg_id (str) – The message for the header

Return type:

Dict[str, str]

Returns:

Header for the AppLogAdapter

sl.snippet_api.snippet.getAppLogger(app_id=None, did=None, msg_id=None, logger=None)

Configure a logger for snippets.

Given a snippet details, create a logging adapter with a header that makes it easy to grep the logs.

Parameters:

app_id (Optional[str]) – A snippet app ID.
did (Optional[int]) – The device ID.
msg_id (Optional[str]) – A string that will identify messages from this logging instance.
logger (Optional[Logger]) – A pre-defined logger to wrap instead of the default silo namespace logger. Most commonly this will be a silo_logger to add context to log messages going to silo.log.

Return type:

Logger

Returns:

A logger adapter with extra content headers.

sl.snippet_api.snippet.getLogger(name=None)

Helper to get a silo namespace logger.

Passing a logger name that already includes the silo namespace does nothing. Passing a logger name that does not include the silo namespace will prepend it.

This is helpful to use in a library that is not in the silo namespace but will be eventually.

Parameters:: name (Optional[str]) – Name of the logger to prepend
Return type:: Logger
Returns:: The logger object

sl.snippet_api.snippet.get_device_logger(did, app=None, dbc=None)

Configure a logger used to write device messages

Parameters:

did (int) – Device ID for the message
app (Optional[object]) – Dynamic App object (self in DA context)
dbc (Union[silo_cursor, SiloCursor, None]) – The database cursor. If none is provided, a local database cursor will be used

Return type:

Logger

Returns:

The configured device logger

sl.snippet_api.snippet.get_index_max_length()

Determine the max length of an index

Return type:: int
Returns:: The maximum character length of an index

sl.snippet_api.snippet.get_model_type()

Lookup the local model_type of the appliance

Refer to sl.snippet_api.platform_definitions.txt_silo_model for model types.

Raises:: MalformedConfig – Configuration error exception
Return type:: int
Returns:: Model Type

sl.snippet_api.snippet.get_module_id(dbc=None)

Lookup the module ID from the licenses table.

Lookup the current module_id for the given database cursor. If a database cursor is not supplied a local to the current instance will be attempted.

Parameters:: dbc (Union[silo_cursor, SiloCursor, None]) – The database cursor. If none is provided, a local database cursor will be used
Return type:: int
Returns:: Module ID

sl.snippet_api.snippet.get_process_info(filename=None, sl1_id=None, dbc=None)

Lookup information for a SL1 process

Must specify filename or sl1_id but not both; otherwise, a ValueError will be raised.

Parameters:

filename (Optional[str]) – Filename to use when searching
sl1_id (Optional[int]) – SL1 id for the process
dbc (Union[silo_cursor, SiloCursor, None]) – The database cursor. If none is provided, a local database cursor will be used

Raises:

ValueError – input or output exception

Return type:

dict

Returns:

SL1 process information

sl.snippet_api.snippet.get_root_logdir()

Determine the root logging directory

Return type:: str
Returns:: File-path of the root logging directory

sl.snippet_api.snippet.get_sl1_version()

Determine the version of SL1

Return type:: Optional[str]
Returns:: The version of the SL1 system

sl.snippet_api.snippet.get_universal_cred_name(cred_id)

Lookup the Universal Credentials Type Name

Parameters:: cred_id (int) – ID of the Credential to be used for universal credential lookups
Return type:: Optional[str]
Returns:: Universal Credential Type Name

sl.snippet_api.snippet.get_universal_protected_fields(cred_id)

Get the list of protected fields for a UCF

Relies on the credential field for encrypted types to take the following form in the collector_state.V_credential_universal table.

"secret_field":
    {
        "type":"Buffer",
        "data": encrypted_data
    }

Protected fields are understood be to encrypted fields.

Parameters:: cred_id (Union[int, str]) – ID of the credential
Return type:: list
Returns:: The list of protected fields

sl.snippet_api.snippet.is_sl1_debug_enabled()

Determines if the process is running in debug mode

Return type:: bool
Returns:: True if the process is running in debug mode, False otherwise

sl.snippet_api.snippet.is_sl1_env()

Determine if the process is running on a SL1 environment

Return type:: bool
Returns:: True if running on a SL1 environment, False otherwise

sl.snippet_api.snippet.is_sl1_interactive_enabled()

Determines if the process is running in interactive mode

Return type:: bool
Returns:: True if the process is running in interactive mode (such as UI debug). False otherwise

sl.snippet_api.snippet.log_root_device(app, log_data, msg_device=None)

Generate an event for root device only if the Exception instance is a CredentialError.

Parameters:

app – Snippet DA reference (self)
log_data (dict) – Data generated from generate_log_data
msg_device (Optional[str]) – Override the default device log message. Default is None

Return type:

None

sl.snippet_api.snippet.normalize_unicode(st)

Normalize unicode code points to ascii to make strings safe for SL1.

Converts unicode strings into ascii strings by substituting ascii equivalents when possible (i.e. n with tilde becomes n), and converting to hex codes in square brackets when substitution is not possible.

For example:

A = ascii char

L = latin1 char with ascii equivalent

La = latin1 char converted to ascii

U = unicode char > 255

U = unconvertible escaped characters

“A L A U” –> “A La A U”

Parameters:: st (str) – string to convert, possibly with unicode code points
Return type:: str
Returns:: ascii string

sl.snippet_api.snippet.platform_versions(*v, **k)

Returns a dict of information about the build of EM7 on this machine

Return type:: Optional[dict]

sl.snippet_api.snippet.silo_conf(filename=None)

Lookup the SL1 configuration information

Parameters:: filename (Optional[str]) – Path to the config file
Return type:: Optional[dict]
Returns:: Dictionary of the parsed config