pykern package¶
pykern package
- copyright:
Copyright (c) 2018 RadiaSoft LLC. All Rights Reserved.
- license:
Subpackages¶
- pykern.api package
- Submodules
- pykern.api.auth_api module
- pykern.api.client module
- pykern.api.server module
- pykern.api.unit_util module
- pykern.api.util module
- pykern.pkcli package
CLI_PKGCommandErrorCustomFormatterCustomParserDEFAULT_COMMANDcommand_error()command_info()main()- Submodules
- pykern.pkcli.ci module
- pykern.pkcli.fmt module
- pykern.pkcli.github module
- pykern.pkcli.github_orgmode module
- pykern.pkcli.pkexample module
- pykern.pkcli.projex module
- pykern.pkcli.rsmanifest module
- pykern.pkcli.sim module
- pykern.pkcli.sphinx module
- pykern.pkcli.test module
- pykern.pkcli.web module
- pykern.pkcli.xlsx module
Submodules¶
pykern.fconf module¶
Python and YAML configuration file parser.
FConf reads Python and YAML files and produces a single, merged PKDict. FConf is not a replacement for pykern.pkconfig. Rather it is for complex configuration input files to programs that often require programmatic generation. FConf was written for RSConf.
The Basic YAML configuration look like this:
fconf_macros:
make_uri(host):
"https://${host}"
some:
- item
uri: make_uri('sirepo.com')
Macros in the fconf_macros are simple replacements. They cannot have loops or other conditions. These are reserved for Python files.
A macro use must be a full text element. It cannot be embedded in
text, e.g. curl make_uri('sirepo.com'), that’s because macros can
return data structures. The syntax is pure Python. Unlike YAML strings,
the arguments have to be quoted, because they will be evaluated by the Python
interpreter.
More complex macros can be programmed in a Python file:
import math
def hypotenuse(self, a, b):
return math.sqrt(a ** 2, b ** 2)
Using in YAML looks the same:
long_side: hypotenuse(3, 4)
Note that the arguments were not quoted here, because they are integers.
The typical usage is:
fconf.parse_all(some_dir)
where some_dir is a py.path that contains python and yaml files to parse. You can also call Parser directly with the a list of files to parse.
As noted, the arguments to a macro are Python. Actually, the entire value is a Python expression, and the only requirement is that it begin with a Python word, which can be any Python function. For example:
count_down:
- reversed(range(4))
Produces the following Python Structure:
{'count_down': [3, 2, 1, 0]}
This allows for a lot of possibilities within the YAML alone.
The Python and YAML macros can call each other. Inside a Python file, just use self to call something:
def count_up(self, top):
return reversed(self.count_down(top))
Note that the above count_down example used a special feature of result merging. When a Macro returns a list (generator, tuple, etc.) in a list context, it will merge with the list. Strings are used in place. For example:
mixed:
- range(3)
- make_uri('sirepo.com')
Produces:
{'mixed': [0, 1, 2, 'https://sirepo.com']}
The same is true of dicts. For example:
fconf_macros:
base(num):
v${num}.radia.run:
description: VM ${num}
all_hosts:
base(3):
base(5):
Produces:
{
"all_hosts": {
"v3.radia.run": {
"description": "VM 3",
},
"v5.radia.run": {
"description": "VM 5",
},
},
}
There are some other useful features. You can refer to any previous declared value using variable expansion. For example:
a:
b:
c: 3
d: ${a}
e: ${a.b.c}
Produces:
{
'a': {'b': {'c': 3}},
'd': {'b': {'c': 3}},
'e': 3,
}
This allows for flexible constants. Those global strings can be used in YAML Macros.
YAML files are evaluated before they are merged. However, they are all merged before the next file is evaluated. This allows a main “constants” file, for example, to direct the flow of the subsequent files.
Workarounds¶
Most strings do not need to be quoted in YAML, but with the extra syntax of FConf, there are some special cases, for example, this fails:
a: 1
b: [ ${a} ]
It gets the error ruamel.yaml.parser.ParserError: while parsing a
flow sequence in "<unicode string> did not find expected ',' or
']'. To work around, simply put in quotes:
a: 1
b: [ "${a}" ]
Another case is inline Python with braces like this:
a: c({"d": "e"})
Just quote with single quotes:
a: 'c({"d": "e"})'
- copyright:
Copyright (c) 2022 RadiaSoft LLC. All Rights Reserved.
- license:
pykern.mpi module¶
MPI support routines
- copyright:
Copyright (c) 2017 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.mpi.checked_call(op)[source]¶
Abort MPI if op raises an exception.
If
opdoesn’t handle an exception, then MPI needs to abort. This will terminate the MPI process, not just the one task.If MPI is not running or mpi4py is not installed, then passes through the exception.
- Parameters:
op (func) – function to call
pykern.pkarray module¶
Wrapper for array to simplify and make future compatible.
Not a complete wrapper. New routines added as required.
- copyright:
Copyright (c) 2015 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkarray.DOUBLE_TYPECODE = 'd'¶
Future-proof typecode for double
- pykern.pkarray.FLOAT_TYPECODE = 'f'¶
Future-proof typecode for float
- pykern.pkarray.new_double(*args, **kwargs)[source]¶
Creates a new double (“d”) array
Args are the same as
array.array()except for typecode, which is passed by this module.- Returns:
New, initialized array
- Return type:
pykern.pkasyncio module¶
Utilities to support asyncio and to simplify starting tornado.
See also pkykern.api which provides a higher level mechanism for websocket clients and servers.
- copyright:
Copyright (c) 2022-2025 RadiaSoft LLC. All Rights Reserved.
- license:
- class pykern.pkasyncio.ActionLoop[source]¶
Bases:
objectProcesses actions in a loop fed by a queue.Queue running in threading.Thread.
Useful to avoid blocking main loop in asyncio for compute or blocking libraries that do not support asyncio.
Calling action is asyncio and thread safe. If the action methods transfer data back to the asyncio loop, use asyncio.call_soon_threadsafe on the appropriate asyncio loop. See discussion in action.
- action(method, arg)[source]¶
Queue
methodto be called in loop thread.Actions are methods that (by convention) begin with
action_and are called sequentially inside _start. A lock is used to prevent destroy being called during the action and serializing activities within a single action.Actions return
Noneto continue on to the next action. _LOOP_END should be returned to terminate _start (the loop) in which case no further actions are performed. Actions can return a callable that will be called inside the loop and outside the lock. These returned callables are known as external callbacks, that is, functions that may do anything so holding the lock could be problematic.The lock is managed by this class and subclasses should not need locking. Resources should be “handed off” to actions via arg passed to method, which can “return” the resource by returning a callback that gets called outside the lock but within the single thread of control that an ActionLoop represents. Read the Go Channels Tutoral for more information about using queues for resource sharing without locks.
- class pykern.pkasyncio.Loop[source]¶
Bases:
objectHTTP Server loop
- http_server(http_cfg)[source]¶
Instantiate a tornado web server
Under the covers Tornado uses the asyncio event loop so asyncio methods can be mixed with Tornado methods.
Using asyncio methods, e.g. asyncio.run, is preferred over Tornado methods, e.g. tornado.ioloop.IOLoop.current to reduce dependency on Tornado. Using this module should allow the code to be portable to other http server frameworks.
http_config.uri_mapmaps URI expressions to classes, which is passed directly to tornado.web.Application.- Parameters:
http_cfg (PKDict) – quest_start, uri_map, debug, tcp_ip, tcp_port,
- pykern.pkasyncio.create_task(coro)[source]¶
Create a task
Keeps a global reference to the task so to avoid the garbage collector running before the task is run. https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task
pykern.pkcollections module¶
PKDict abstraction and utils
PKDict is similar to argparse.Namespace, but is a dict that allows
you to treat elements as attributes.
- copyright:
Copyright (c) 2015-2022 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkcollections.DictNameError¶
alias of
PKDictNameError
- class pykern.pkcollections.PKDict[source]¶
Bases:
dictA subclass of dict that allows items to be read/written as attributes.
The purpose of this is as a convenience in coding. You can refer to dictionary keys as attributes as long as they don’t collide with the object’s attributes. You should always use the dict interface to refer to the items of the dictionary programmatically, just like you would do with Javascript.
You can reference any dict key as an attribute as long as it does not conflict with an attribute. For example, this works:
x = PKDict() x.a = 1 assert 1 == x.a x['a'] = 3 assert 3 == x.a assert 'a' == x.keys()[0]
You can’t set an attribute that already exists. These calls throw exceptions:
x.values = 1 delattr(x, 'values')
dict doesn’t allow this anyway. However, you can’t set or delete any existing attribute, even writable attributes. Indeed, you can’t delete attributes at all. Subclasses should be “containers” only, not general objects.
- copy()[source]¶
Override dict.copy to ensure the class of the return object is correct.
Necessary because dict.copy will return a dict rather than PKDict. Calls self.__class__(self) which makes a shallow copy. Subclasses that do not accept this constructor form will need to override this method.
- Returns:
shallow copy of self
- Return type:
- pkdel(name, default=None)[source]¶
Delete item if exists and return value
The code will survive against concurrent access, but is not thread safe.
Never throws KeyError.
- pkmerge(to_merge, make_copy=True)[source]¶
Add to_merge to self
Types are assumed to match and are not converted, e.g. dict is not converted to PKDict. Again, use canonicalize if you want to avoid type incompatibilities.
to_merge’s values override self’s so if say, to_merge is
{'x': None}, thenself.xwill be None at the end of this call even if it had a value before.Lists from to_merge are prepended on this same principle, that is, to_merge “overrides” self, and prepending is defined as overriding. Lists must contain unique elements and duplicates will cause an error.
This function recurses only on PKDicts.
- pknested_get(qualifiers)[source]¶
Split key on dots or iterable and return nested get calls
If qualifiers is a str, will split on dots. Otherwise, will be iterated.
If an element is a list or tuple, tries to index as int.
Throws KeyError if the dictionary key doesn’t exist.
- pknested_set(qualifiers, value)[source]¶
Set nested location identified by qualifiers to value
If qualifiers is a str, will split on dots. Otherwise, will be iterated.
If an element does not exist, creates it as a PKDict.
DOES NOT SUPPORT ints (lists) at this time.
- pksetdefault(*args, **kwargs)[source]¶
Set values for keys if key not already in self
Accepts args or kwargs, but not both. args must be an even number and are interpreted in (key, value) order. kwargs are accepted as key=value.
If self does not have key, then it will be set to value. If key is already in self, its value is not changed. If value is a callable, it will be called, and key will be set to the returned value.
- pksetdefault1(*args, **kwargs)[source]¶
Get value if exists else set
Accepts one key and one value, either as two positional args (
key, value); or one keyword arg (key=value).If self does not have key, then it will be set to value. If value is a callable, it will be called, and key will be set to the returned value.
- pkunchecked_nested_get(qualifiers, default=None)[source]¶
Split key on dots or iterable and return nested get calls
If qualifiers is a str, will split on dots. Otherwise, will be iterated.
If the element does not exist or is not indexable, fails silently and returns default.
Throws KeyError if the dictionary key doesn’t exist.
- exception pykern.pkcollections.PKDictNameError[source]¶
Bases:
NameErrorRaised when a key matches a builtin attribute in dict.
- pykern.pkcollections.canonicalize(obj)[source]¶
Convert to lists and PKDicts for simpler serialization
Traverse obj to convert all values to forms that are compatible with serialization protocols like YAML or JSON.
Simple objects are ensured to match their types e.g. bool, float, int, and str. Objects that are instances of these, are converted to these to ensure they are basic types, that is,
canonicalize(str_subclass('a'))will be converted tostr('a').bytes and bytearrays will be converted to str.
decimal.Decimal will converted to float.
All objects are traversed. If no objects need to be converted, obj will be returned unmodified.
Generators and iterables are converted to lists.
Circularities are not detected so infinite recursion can occur.
pykern.pkcompat module¶
Backwards and forward compatible Python utilities
Functions here will be available forever, even when functions are removed from Python.
Some functions are no longer necessary, e.g. unicode_getcwd. Others are still convenient, e.g. from_bytes is useful when you don’t know whether it is a str or not.
- copyright:
Copyright (c) 2015-2023 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkcompat.from_bytes(value)[source]¶
Converts value to a str
If value is not a str, decode with utf-8.
If already str, does nothing.
- pykern.pkcompat.locale_str(value)[source]¶
Converts a value to a unicode str unless already unicode.
- pykern.pkcompat.to_bytes(value)[source]¶
Converts a value to bytes
If value is a str, encode with utf-8.
If already bytes, does nothing.
- pykern.pkcompat.unicode_getcwd()[source]¶
os.getcwd()unicode wrapper- Returns:
current directory
- Return type:
- pykern.pkcompat.unicode_unescape(value)[source]¶
Convert escaped unicode and Python backslash values in str
pykern.pkconfig module¶
Declarative module configuration with dynamic value injection
Module Declaration¶
Modules declare their configuration via init. Here is how pkdebug declares its config params:
cfg = pkconfig.init(
control=(None, re.compile, 'Pattern to match against pkdc messages'),
want_pid_time=(False, bool, 'Display pid and time in messages'),
output=(None, _cfg_output, 'Where to write messages either as a "writable" or file name'),
)
A param tuple contains three values:
Default value, in the expected type
Callable that can convert a string or other type into a valid value
A docstring briefly explaining the configuration element
The returned cfg object is ready to use after the call. It will contain
the config params as defined or an exception will be raised.
Config Values¶
Configuration is returned as nested dicts. The values themselves could be any Python object. In this case, we have a string and a file object for the two parameters. We called os.getcwd and referred to sys.stdout in param values.
Summary¶
Here are the steps to configuring an application:
When the first module calls init, pkconfig gets environment variables to create a single dict of param values, unparsed.
init looks for the module’s params in the unparsed values.
If the parameter is found, that value is used. Else, the default is merged into the dict and used.
The parameter value is then resolved with str.format. If the value is a list it will be joined with any previous value (e.g. default).
The resolved value is parsed using the param’s declared
parser.The result is stored in the merged config and also stored in the module’s Params object .
Once all params have been parsed for the module, init returns the Params object to the module, which can then use those params to initialize itself.
- copyright:
Copyright (c) 2015-2019 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkconfig.CHANNEL_ATTR = 'pykern_pkconfig_channel'¶
Environment variable holding channel (defaults to “dev”)
- pykern.pkconfig.CHANNEL_DEFAULT = 'dev'¶
Initialized channel (same as cfg.channel)
- pykern.pkconfig.INTERNAL_TEST_CHANNELS = ('dev', 'alpha')¶
Channels which can have more verbose output from the server
- pykern.pkconfig.KEY_RE = re.compile('^[A-Z][A-Z0-9_]*[A-Z0-9]$', re.IGNORECASE)¶
Cannot begin with non-letter or end with an underscore
- Type:
Validate key
- class pykern.pkconfig.ReplacedBy(new_name)[source]¶
-
Container for a required parameter declaration.
Example:
cfg = pkconfig.init( gone=pkconfig.ReplacedBy('pykern.pkexample.foo'), )
- Parameters:
new_name – name of new config parameter
- class pykern.pkconfig.Required(*args)[source]¶
-
Container for a required parameter declaration.
Example:
cfg = pkconfig.init( any_param=(1, int, 'A parameter with a default'), needed=pkconfig.Required(int, 'A parameter without a default'), )
- Parameters:
converter (callable) – how to string to internal value
docstring (str) – description of parameter
- class pykern.pkconfig.RequiredUnlessDev(*args)[source]¶
-
Container for a required parameter declaration only in_dev_mode.
Only required if in_dev_mode is true.
Example:
cfg = pkconfig.init( maybe_needed=pkconfig.RequiredUnlessDev('dev default', str, 'A parameter with a default'), )
- pykern.pkconfig.STRING_TYPES¶
Python version independent value of string instance check
- pykern.pkconfig.TUPLE_SEP = ':'¶
parse_tuple splits strings on this
- pykern.pkconfig.VALID_CHANNELS = ('dev', 'alpha', 'beta', 'prod')¶
Order of channels from least to most stable
- pykern.pkconfig.cfg = {'channel': 'dev', 'dev_mode': True}¶
channel
- Type:
Configuration for this module
- pykern.pkconfig.flatten_values(base, new)[source]¶
Merge flattened values into base
Keys are made all lowercase.
Lists instances are prepended and not recursively merged.
- pykern.pkconfig.in_dev_mode()[source]¶
Turn on developer features
- Returns:
value of pykern.pykern.cfg.dev_mode
- Return type:
- pykern.pkconfig.init(**kwargs)[source]¶
Declares and initializes config params for calling module.
- Parameters:
kwargs (dict) – param name to (default, parser, docstring)
- Returns:
PKDict populated with param values
- Return type:
Params
- pykern.pkconfig.parse_bool(value)[source]¶
Default parser for bool types
When the parser is bool, it will be replaced with this routine, which parses strings and None specially. bool values cannot be defaulted to None. They must be True or False.
String values which return true: t, true, y, yes, 1.
False values: f, false, n, no, 0, ‘’, None
Other values are parsed by bool.
- pykern.pkconfig.parse_none(func)[source]¶
Decorator for a parser which can parse None
- Parameters:
callable – function to be decorated
- Returns:
func with attr indicating it can parse None
- Return type:
callable
- pykern.pkconfig.parse_secs(value)¶
deprecated version of parse_seconds
- pykern.pkconfig.parse_set(value)[source]¶
Default parser for set and frozenset types
When the parser is set or frozenset, it will be replaced with this routine, which parses strings, lists, and sets. It splits strings on ‘:’.
- pykern.pkconfig.parse_tuple(value)[source]¶
Default parser for tuple types
When the parser is tuple, it will be replaced with this routine, which parses strings, lists, sets, and tuples. It splits strings on ‘:’.
- pykern.pkconfig.reset_state_for_testing(add_to_environ=None)[source]¶
Clear the raw values and append add_to_environ
Only used for unit tests.
add_to_environoverrides previous value.- Parameters:
add_to_environ (dict) – values to augment to os.environ
- pykern.pkconfig.to_environ(cfg_keys, values=None, exclude_re=None)[source]¶
Export config (key, values) as dict for environ
cfg_keys is a list of dotted words (
['pykern.pkdebug.control']).Simple globs (
pykern.pkdebug.*) are supported, which match any word character. This can be used to ensure there is enough depth, e.g. pykern.*.* means there must be at least two undercores in the environment variable.Only environ and add_to_environ config will be considered, not default values, which will be assumed to be processed the same way in a subprocess using this environ.
pykern.pkconst module¶
Various constant values
- copyright:
Copyright (c) 2019 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkconst.BASE62_CHARS = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'¶
Copied from numconv, which copied from RFC1924
- pykern.pkconst.LOCALHOST_IP = '127.0.0.1'¶
IP address for local servers
- pykern.pkconst.PACKAGE_DATA = 'package_data'¶
The subdirectory in the top-level Python where to put resources
- pykern.pkconst.PY_PATH_LOCAL_TYPE¶
Class of pkio.py_path and py.path.local
- pykern.pkconst.STRING_TYPES¶
DEPRECATED use str
- pykern.pkconst.builtin_print(*args, sep=' ', end='\n', file=None, flush=False)¶
Use this (sparingly, pkdlog is prefered) when you want to use print directly.
pykern.pkdebug module¶
Logging or regex-controlled print statements
We take the view that there are two types of logging: permanent (pkdp) and dynamically controlled (pkdc). In both cases, you want to know where the print statement originated from. Python’s logging module supports logging the origin, but it doesn’t support user- defined dynamic filters in the form of a regular expression.
Permanent print statements (pkdp) is what people traditionally think of as “logging”. What we don’t care about is “logging levels”, because one person’s “critical” is another person’s “whatever”. It’s up to operations people to figure out what’s an error worth paying attention to, and what’s not. That’s managed by external rule bases, not inside some library module.
Python’s logging module gives programmers too many options and adds complexity to filters. If you don’t attach a handler to the loggers, you’ll never see some messages. Or, if the handler has logging.WARN set, and you really want to see logging.INFO messages, you won’t see them.
There are cases when you really don’t want to see output. We call these dynamically controlled print statements. They are typically used by developers when debugging a new module, and sometimes these lines get left in the code for a while so that you can get detailed (noisy) debugging information on production systems.
Dynamically controlled print statements (pkdc) would overwhelm logs so they are off by default and can be filtered by regular expressions supplied via configuration (including environment variables) or programmatically via init.
Examples
In a module, you would write:
from pykern.debug import pkdc, pkdexc, pkdp
pkdp('user entered: {}', val)
pkdc('user context: name={name}, id={id}', **user_rec)
The first print statement always appears.
The second does not unless you specify a “control” via the
environment variable $PYKERN_PKDEBUG_CONTROL:
PYKERN_PKDEBUG_CONTROL=my_mod python my_prog.py
or with the shortname pkdebug:
pkdebug=my_mod python my_prog.py
Or, if you want a specific conditional print:
PYKERN_PKDEBUG_CONTROL=my_mod.py:52:
You can match any text in the line output with a regular expression, which is case insensitive.
If you omit args and kwargs, pkdp returns its argument, of use when you want to inspect values within the normal flow of the code, e.g.:
def my_function():
...
return pkdp(res)
or
a = pkdp(my_function_1()) + pkdp(my_function_2())
NOTE: format arguments are converted to strings by this module, and not by str.format. Then they are passed to str.format. This means that you should assume the argument is a string or, better, just a default type. Arguments are trunctated by pkdformat.
If output is a string, will open the file to write to. The initial
value of output is $PYKERN_PKDEBUG_OUTPUT.
- copyright:
Copyright (c) 2014-2016 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkdebug.MAX_EXCEPTION_COUNT = 5¶
Maximum number of exceptions thrown before printing stops
- pykern.pkdebug.REDACTED = '<REDACTED>'¶
redeacted value marker
- pykern.pkdebug.SECRETS_RE = re.compile('(?:secret|otp\\b|passw|private)', re.IGNORECASE)¶
keys whose values will be redacted from logs
- pykern.pkdebug.SNIP = '<SNIP>'¶
truncated string marker
- pykern.pkdebug.init(**kwargs)[source]¶
May be called to (re)initialize this module.
control is a regular expression, which is used to control the output of
pkdc(). Messages frompkdp()andpkdc()are written to output.output is either an object which implements write or a str, in which case it is opened with
io.open().
- pykern.pkdebug.pkdc(fmt, *args, **kwargs)[source]¶
Conditional print a message to output selectively based on control.
Controlled output that you can leave in your code permanently.
- Parameters:
fmt (str) – how to
str.format()args – what to format
kwargs – what to format
- pykern.pkdebug.pkdexc(exc_info=None, simplify=False)[source]¶
Return last exception and stack as a string
Must be called from an
except. This function removes the last two calls from the stack. It joins the traceback so you get a complete stack trace.Will catch exceptions during the formatting and returns a string in all cases.
Example:
try: something except: pkdlog("exception={}", pkdexc())
- pykern.pkdebug.pkdformat(fmt, *args, **kwargs)[source]¶
Truncate and redact args & kwargs and then format
NOTE: converts all elements to strings before calling str.format.
- pykern.pkdebug.pkdlog(fmt_or_arg, *args, **kwargs)[source]¶
Print a message to output unconditionally, possibly returning its arg
Use for print statements in your code that you don’t intend to keep in the system.
Use pkdlog for permanent log messages.
- pykern.pkdebug.pkdp(fmt_or_arg, *args, **kwargs)[source]¶
Print a message to output unconditionally, possibly returning its arg
Use for print statements in your code that you don’t intend to keep in the system.
Use pkdlog for permanent log messages.
pykern.pkexample module¶
Demonstrates a RadiaSoft style module.
This module demonstrates how we code at RadiaSoft. In general we follow the Google Python Style Guide and the Sphinx Napoleon Google Docstrings Example.
- Some Rules:
We adhere to PEP8 unless overruled by Google’s Style Guide or explicit rules such as:
Indent by four (4) spaces and no tabs.
Avoid lines over 80 characters. Not everybody’s laptop can show two 80+ char pages side by side.
All modules are Python 2 and 3 compatible.
Modules are divided into groups of declarations: public constants, private constants, public global variables, private global variables, public classes, private classes, public functions, and private functions. Within each group, the declarations are sorted alphabetically.
Logs, exceptions and assertions contain values, not just messages. If a value is in error, include it in the message along with the expected value, range, etc.
:func:pykern.pkdebug.pkdp to print logging messages, and :func:pykern.pkdebug.pkdc to print trace messages.
Configuration is specified by :mod:pykern.pkconfig.
Use single quotes for strings. Use double quotes for docstrings.
String literals should be concated with
+not by implicit concatenationTODO(robnagler) is how we represent things to do in a comment
Docstrings begin and end with three double quotes (“). On the line with the beginning double quotes, you write a one line summary of the function, class, or module.
- copyright:
Copyright (c) 2016 RadiaSoft LLC. All Rights Reserved.
- license:
- class pykern.pkexample.EMA(length)[source]¶
Bases:
objectExponential moving average
Used as a demonstration of numerical computations. We document the __init__ method at the class level, since it is an implicit function call.
- Parameters:
length (int) – iterations
- pykern.pkexample.ONE = 1¶
First constant is first alphabetically
- pykern.pkexample.ZIPPITY = 'doodah'¶
Another constant
- pykern.pkexample.cfg = {'any_length': 1}¶
The module’s config is a public variable. It is initialized at the end.
pykern.pkinspect module¶
Helper functions for to inspect.
- copyright:
Copyright (c) 2015 RadiaSoft, Inc. All Rights Reserved.
- license:
- class pykern.pkinspect.Call(frame_or_log)[source]¶
Bases:
PKDictSaves file:line:name of stack frame and renders as string.
- Parameters:
frame_or_log (frame or LogRecord) – values to extract
- exception pykern.pkinspect.SubmoduleNotFound[source]¶
Bases:
ModuleNotFoundErrorRaised by import_submodule
- pykern.pkinspect.append_exception_reason(exc, reason)[source]¶
Augment exc with reason
Modifies exc in place by adding to exc.args or exc.reason. Does it’s best to not cause another exception during this process.
- Parameters:
exc (BaseException) – what was raised
reason (str) – our related reason
- pykern.pkinspect.caller(ignore_modules=None, exclude_first=True)[source]¶
Which file:line:func is calling the caller of this function.
Will not return the same module as the calling module, that is, will iterate until a new module is found. If ignore_modules is defined, will ignore those modules as well.
Note: may return __main__ module.
Will raise exception if calling from __main__
- pykern.pkinspect.caller_func_name()[source]¶
Name of function one frame back
Useful for inter-module dispatch and errors.
- Returns:
function name
- Return type:
- pykern.pkinspect.caller_module(exclude_first=True)[source]¶
Which module is calling the caller of this function.
Will not return the same module as the calling module, that is, will iterate until a new module is found. If exclude_first == True it will also exclude the first module found that is not the calling module.
Note: may return __main__ module.
Will raise exception if calling from __main__
- Parameters:
exclude_first (bool) – skip first module found [True]
- Returns:
module which is calling module
- Return type:
module
- pykern.pkinspect.import_submodule(submodule, subpackage=None, root_packages=None)[source]¶
Import a module of the form
root.subpackage.submoduleSearch
root_packages, e.g.(sirepo, pykern), for modules within the a subpackage of the roots.- Parameters:
- Returns:
imported module object
- Return type:
module
- Raises:
SubmoduleNotFound – for submodule not being found vs ModuleNotFoundError when any module imported by submodule is in error.
- pykern.pkinspect.is_caller_main()[source]¶
Is the caller’s calling module __main__?
- Returns:
True if calling module was called by __main__.
- Return type:
- pykern.pkinspect.module_basename(obj)[source]¶
Parse the last part of a module name
For example, module_basename(pkinspect) is ‘pkinspect’.
- pykern.pkinspect.module_functions(func_prefix, module=None)[source]¶
Get all module level functions starting with func_prefix
- pykern.pkinspect.module_name_join(names)[source]¶
Joins names with ‘.’
- Parameters:
names (iterable) – list of strings to join
- Returns:
module name
- Return type:
- pykern.pkinspect.package_module_names(name_or_module)[source]¶
List of modules of package name_or_module
- pykern.pkinspect.root_package(obj)[source]¶
Parse the root package in which obj is defined.
For example, root_package(module_basename) is ‘pykern’.
pykern.pkio module¶
Useful I/O operations
- copyright:
Copyright (c) 2015 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkio.atomic_write(path, contents=None, writer=None, **kwargs)[source]¶
Overwrites an existing file with contents via rename to ensure integrity
- pykern.pkio.compare_files(path1, path2, force=False)[source]¶
Compares two files using filecmp.cmp
Note that filecmp uses os.stat to see if a file is the same. If the size, mtime, and type are not identical, it does a comparison of the contents.
filecmp caches prior resuls of the content comparisons. force “ensures” no caching, but since the cache is global, this can’t be guaranteed in multithreaded environments.
- pykern.pkio.exception_is_not_found(exc)[source]¶
True if exception is one various file not found errors
Checks FileNotFoundError and IOError with errno.ENOENT.
- Parameters:
exc (BaseException) – to check
- Returns:
True if is a file not found exception.
- Return type:
- pykern.pkio.is_pure_text(filepath, test_size=512)[source]¶
Read test_size bytes of filepath to determine if it is likely a text file.
See pykern.util.is_pure_text for the heuristics used to test bytes.
- pykern.pkio.mkdir_parent(path)[source]¶
Create the directories and their parents (if necessary)
- Parameters:
path (str) – dir to create
- Returns:
path
- Return type:
py.path.local
- pykern.pkio.mkdir_parent_only(path)[source]¶
Create the paths’ parent directories.
- Parameters:
path (str) – children of dir to create
- Returns:
parent directory of path
- Return type:
py.path.local
- pykern.pkio.pkunit_prefix = None¶
used during unit testing see
pykern.pkunit.save_chdir
- pykern.pkio.py_path(path=None)[source]¶
Creates a py.path.Local object
Will expanduser, if needed.
If pkunit_prefix is set, will prefix, too.
- Parameters:
path (str) – path to convert (or None for current dir)
- Returns:
path
- Return type:
py.path.Local
- pykern.pkio.save_chdir(dirname, mkdir=False, is_pkunit_prefix=False)[source]¶
Save current directory, change to directory, and restore.
- pykern.pkio.sorted_glob(pattern, key=None)[source]¶
Returns sorted list of files & dirs matching pattern.
Use ‘**’ in pattern for recursive search, else, use * as wildcard. Sorts using key if provided, else in ascending order. Doesn’t include dot files unless dot “.” is included explicitly at the start of a path component. Doesn’t include . and .. To return files only, see walk_tree().
- pykern.pkio.unchecked_remove(*paths)[source]¶
Remove files or directories, ignoring OSError.
Will not remove ‘/’ or ‘.’
- Parameters:
paths (str) – paths to remove
- pykern.pkio.walk_tree(dirname, file_re=None)[source]¶
Returns list of all files (only) in dirname (recursive), sorted in ascending order.
Include file_re to filter results. Includes dot files, but not . and .. To include dirs in results, see sorted_glob().
pykern.pkjinja module¶
Simplify rendering jinja2
- copyright:
Copyright (c) 2015 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkjinja.RESOURCE_SUFFIX = '.jinja'¶
Implicit extension including ‘.’ added to resources
pykern.pkjson module¶
JSON wrapper
- copyright:
Copyright (c) 2017-2023 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkjson.CONTENT_TYPE = 'application/json; charset="utf-8"'¶
Content-Type MIME header
- pykern.pkjson.ENCODING = 'utf-8'¶
how bytes are encoded
- class pykern.pkjson.Encoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]¶
Bases:
JSONEncoder- default(obj)[source]¶
Implement this method in a subclass such that it returns a serializable object for
o, or calls the base implementation (to raise aTypeError).For example, to support arbitrary iterators, you could implement default like this:
def default(self, o): try: iterable = iter(o) except TypeError: pass else: return list(iterable) # Let the base class default method raise the TypeError return super().default(o)
- pykern.pkjson.MIME_TYPE = 'application/json'¶
MIME type
- pykern.pkjson.dump_pretty(obj, filename=None, pretty=True, **kwargs)[source]¶
Formats as json as string
If an object is not encoded by default, will call str() on the object.
pykern.pkplatform module¶
Wrapper for Python’s platform to provide cleaner programmatic
control of system features.
- copyright:
Copyright (c) 2015 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkplatform.is_darwin()[source]¶
All flavors of Mac (OS X)
- Returns:
True if
sys.platformis Mac.- Return type:
- pykern.pkplatform.is_linux()[source]¶
All flavors of linux
- Returns:
True if
sys.platformis Linux.- Return type:
pykern.pkresource module¶
Where external resources are stored
- copyright:
Copyright (c) 2015 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkresource.file_path(relative_filename, caller_context=None, packages=None)[source]¶
Return the path to the resource
- pykern.pkresource.filename(relative_filename, caller_context=None, packages=None)[source]¶
Return the filename to the resource
pykern.pkrunpy module¶
Run python code
- copyright:
Copyright (c) 2015 RadiaSoft LLC. All Rights Reserved.
- license:
pykern.pksetup module¶
Wrapper for setuptools.setup to simplify creating of setup.py files.
Python setup.py files should be short for well-structured projects. b_setup.setup assumes there are directories such as tests, docs, bin, etc. PyKern Projects use py.test so the appropriate Test class is provided by this module.
Example
A sample setup.py script:
setup(
name='pyexample',
description='Some Example app',
author='Example, Inc.',
author_email='somebody@example.com',
url='http://example.com',
)
Assumptions:
GUI and console scripts are found automatically by special suffixes
_gui.pyand_console.py. Seesetupdocumentation for an example.Under git control. Even if you are building an app for the first time, you should create the repo first. Does not assume anything about the remote (i.e. need not be a GitHub repo).
- copyright:
Copyright (c) 2015 Radiasoft LLC. All Rights Reserved.
- license:
- pykern.pksetup.PACKAGE_DATA = 'package_data'¶
The subdirectory in the top-level Python where to put resources
- pykern.pksetup.SCRIPTS_DIR = 'scripts'¶
Where scripts live, you probably don’t want this
- class pykern.pksetup.SDist(dist: Distribution, **kw)[source]¶
Bases:
sdist,objectFix up a few things before running sdist
- pykern.pksetup.install_requires()[source]¶
Parse requirements.txt.
- Returns:
parsed requirements.txt
- Return type:
- pykern.pksetup.setup(**kwargs)[source]¶
Parses README.* and requirements.txt, sets some defaults, then calls setuptools.setup.
Scripts are found by looking for files in the top level package directory which end with
_console.pyor_gui.py. These files must have a function calledmain.Example
The file
pykern_console.pymight contain:def main(): return 2 + 2
This would create a program called command line program
pykernwhich would callmain()when invoked.- Parameters:
kwargs – see setuptools.setup
pykern.pksubprocess module¶
Wrapper for subprocess.
- copyright:
Copyright (c) 2016 RadiaSoft LLC. All Rights Reserved.
- license:
pykern.pkunit module¶
Useful operations for unit tests
- copyright:
Copyright (c) 2015 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkunit.DATA_DIR_SUFFIX = '_data'¶
Where persistent input files are stored (test_base_name_data)
- class pykern.pkunit.ExceptToFile[source]¶
Bases:
objectWrites exception or None to PKEXCEPT_PATH
Used for deviance testing with case_dirs.
If there is an exception, writes that to the file. Otherwise, writes “None”
If there is an exception, will write PKSTACK_PATH. Otherwise, no file exists. Used for diagnostics only.
- Usage::
- for d in case_dirs():
- with ExceptToFile():
command to test
- Returns:
just for context manager
- Return type:
None
- pykern.pkunit.LOCALHOST_IP = '127.0.0.1'¶
Used to create test servers
- pykern.pkunit.PKEXCEPT_PATH = 'pkexcept'¶
Where ExceptToFile writes exception
- exception pykern.pkunit.PKFail[source]¶
Bases:
AssertionError
- pykern.pkunit.PKSTACK_PATH = 'pkstack'¶
Where ExceptToFile writes stack
- pykern.pkunit.RESTARTABLE = 'PYKERN_PKUNIT_RESTARTABLE'¶
Environment var set by pykern.pkcli.test if the test is restartable
- pykern.pkunit.TEST_FILE_ENV = 'PYKERN_PKUNIT_TEST_FILE'¶
Environment var set by pykern.pkcli.test for each module under test
- pykern.pkunit.WORK_DIR_SUFFIX = '_work'¶
Where to write temporary files (test_base_name_work)
- class pykern.pkunit.WebServer[source]¶
Bases:
objectServes files from data_dir on a random port in a separate thread.
Usage:
with pkunit.WebServer() as server: # server.url is "http://localhost:<port>" do_something(server.url)
- pykern.pkunit.assert_object_with_json(basename, actual)[source]¶
Converts actual to JSON and compares with data_dir/basename.json
Reads data_dir/basename.json and compares with actual converted to json. Trailing newline is managed properly. The keys are sorted and indentation is 4. actual written to work_dir.
- pykern.pkunit.case_dirs(group_prefix='', **kwargs)[source]¶
Sets up work_dir by iterating
*.inin data_dirEvery
<case-name>.inis copied recursively to<case-name>in work_dir. This function then yields that directory. The test can then run the function to be tested.When test yields to the generator, this function looks for all files in data_dir in the sub-directory
<case-name>.out. Each of these expect files is copmared to the corresponding work_dir actual file using file_eq.If you want to only use cases from some specific <case-name>.in subdir, and not all *.in subdirs, you can pass a group_prefix default parameter value (’’ by default) to case_dirs(). This will perform the regular operations but only on <case-name>.in.
Excel spreadsheets are supported. If the name of the expect (out) file is
foo.csv, then the first sheet (sheet 0) in the corresponding work_dir xlsx will be converted tofoo.csvbefore comparison. If the expect (out) file has a#<digit>, e.g.foo#3.csv, then the fourth sheet will be extracted from the actual xlsx tofoo#3.csvin the work_dir.If ExceptToFile in the body of a case_dirs loop, the exception will be output if a file is not found.
- Parameters:
group_prefix (string) – target subdir [‘’]
j2_ctx (dict) – passed to pykern.pkjinja.render_file
ignore_lines (iterable) – POSIX standard regular expressions to be passed to diff
is_bytes (bool) – do a binary comparison [False]
- Returns:
case directory created in work_dir (also PWD)
- Return type:
py.path.local
- pykern.pkunit.data_dir()[source]¶
Compute the data directory based on the test name
The test data directory is always
<test>_data, where<test>is the name of the test’s python module with the_testortest_removed. For example, if the test file issetup_test.pythen the directory will besetup_data.- Returns:
data directory
- Return type:
py.path.local
- pykern.pkunit.empty_work_dir()[source]¶
Remove work_dir if it exists and create.
All contents of the test directory will be removed.
- Returns:
empty work directory
- Return type:
py.path.local
- pykern.pkunit.file_eq(expect_path, *args, **kwargs)[source]¶
If actual is not expect_path, throw assertion with calling context.
expect_path and actual_path both exist, they will be compared as plain text.
If actual_path does not exist, it will be created from actual.
If expect_path ends in
.jsonand actual_path does not exist, pkjson will be used to load expect_path and a data structure comparison will be used with actual (and actual_path will be written. This allows easy testing of complex results.If expect_path ends with
.jinja, it will be rendered with pykern.pkjina.render_file, and you must supply j2_ctx in kwargs.- Parameters:
expect_path (str or py.path) – text file to be read; if str, then joined with data_dir
actual (object) – string or json data structure; if missing, read actual_path (may be positional)
actual_path (py.path or str) – where to write results; if str, then joined with work_dir; if None,
work_dir().join(expect_path.relto(data_dir()))j2_ctx (dict) – passed to pykern.pkjinja.render_file
ignore_lines (iterable) –
POSIX standard regular expressions to be passed to diff
is_bytes (bool) – do a binary comparison [False]
- pykern.pkunit.import_module_from_data_dir(module_name)[source]¶
Add data_dir to sys.path and import module_name.
Note that module_name with be removed from the sys.modules cache before loading in case the module was loaded by another test.
- Parameters:
module_name (str) – module relative to data_dir to import.
- Returns:
imported module
- Return type:
module
- pykern.pkunit.insert_data_dir_in_sys_path()[source]¶
Context manager to insert data_dir first in sys.path
- pykern.pkunit.is_test_run()[source]¶
Running in a test?
- Returns:
whether this is running in a test
- Return type:
- pykern.pkunit.module_under_test = None¶
Set to the most recent test module by _test_file
- Type:
INTERNAL
- pykern.pkunit.pkeq(expect, actual, *args, **kwargs)[source]¶
If actual is not expect, throw assertion with calling context.
Opposite of pkne.
- pykern.pkunit.pkexcept(exc_or_re, *args, **kwargs)[source]¶
Expect an exception to be thrown and match or output msg
Examples:
# Expect an exception (or its subclass) with pkexcept(AssertionError, 'did not expect this'): assert 0 # Expect exception to contain a specific message with pkexcept('match this', 'problem with matching'): assert 0, 'some string with "match this" in it' # Use a default output message with pkexcept(KeyError): something['key will not be found']
- pykern.pkunit.pkne(expect, actual, *args, **kwargs)[source]¶
If actual is equal to expect, throw assertion with calling context
Opposite of pkeq.
- pykern.pkunit.pkok(cond, fmt, *args, **kwargs)[source]¶
If cond is not true, throw PKFail with calling context
- pykern.pkunit.pkre(expect_re, actual, flags=18)[source]¶
If actual does not match (re.search) expect_re, throw PKFail with calling context.
- pykern.pkunit.restart_or_fail(*args, **kwargs)[source]¶
Test will be restarted (at process level) if it can, else pkfail
Called by tests which experience known CI failures such as not being able to connect to servers.
Communicates with pykern.pkcli.test
- pykern.pkunit.save_chdir_work(is_pkunit_prefix=False, want_empty=True)[source]¶
Change to work_dir which will be created.
Default to empty_work_dir before chdir.
- pykern.pkunit.test_path_to_work_dir(path)[source]¶
Convert a test file path to its work directory path.
Strips
_testsuffix ortest_prefix from the basename and appends_work.- Parameters:
path (str or py.path.local) – test file path ending in
_testor starting withtest_- Returns:
work directory path
- Return type:
py.path.local
- pykern.pkunit.work_dir()[source]¶
Returns ephemeral work directory, created if necessary.
To enable easier debugging, the test directory is always
<test>_work, where<test>is the name of the test’s python module with the_testortest_removed. For example, if the test file issetup_test.pythen the directory will besetup_work.The name “work” distinguishes from “tmp”, which could imply anything. Also, with editor autocomplete, “setup_work” and “setup_test” are more easily distinguishable.
- Returns:
directory name
- Return type:
py.path
pykern.pkyaml module¶
Wrapper for yaml
- copyright:
Copyright (c) 2015-2022 RadiaSoft LLC. All Rights Reserved.
- license:
- pykern.pkyaml.PATH_EXT = '.yml'¶
file extension for yaml
- pykern.pkyaml.dump_pretty(obj, filename=None, pretty=True, ruamel_attrs=None)[source]¶
Formats as yaml as string
If an object is not encoded by default, will call str() on the object.
Unlike pkjson.dump_pretty, returns nothing.
pykern.pykern_console module¶
Front-end command line for pykern.pkcli.
Example:
- copyright:
Copyright (c) 2015 RadiaSoft LLC. All Rights Reserved.
- license:
pykern.pytest_plugin module¶
REMOVED
Left for backward compatibility
- copyright:
Copyright (c) 2016-2024 RadiaSoft LLC. All Rights Reserved.
- license:
pykern.quest module¶
API wrapper
- copyright:
Copyright (c) 2024 RadiaSoft LLC. All Rights Reserved.
- license:
- class pykern.quest.API(*args, **kwargs)[source]¶
Bases:
PKDictHolds request context for all API calls.
- METHOD_PREFIX = 'api_'¶
- class pykern.quest.Attr(qcall, **kwargs)[source]¶
Bases:
PKDict- IS_SINGLETON = False¶
shared Attrs do not have link to qcall
- quest_end(qcall, in_error)[source]¶
Called when quest ends
Right before destroy. No other attributes are available.
pykern.sql_db module¶
EXPERIMENTAL sqlalchemy wrapper
This interface is GOING TO CHANGE.
- copyright:
Copyright (c) 2025 RadiaSoft LLC. All Rights Reserved.
- license:
- exception pykern.sql_db.BaseExc(**context)[source]¶
Bases:
ExceptionSuperclass for sll exceptions in this module
- class pykern.sql_db.Meta(uri, schema)[source]¶
Bases:
objectA practical wrapper for sqlalchemy.engine and session objects.
Schema specification is designed to reduce boilerplate. The general form is:
{table(s): {columns(s): "type modifier(s)", constraint(s): val}.A detailed schema example is in
tests/sql_db_test.py.The type must be one of: bool, datetime, float, int, primary_id, str, text. float and int are followed by a size of 32 or 64. str is followed by a size of any length.
Column modifiers are separated by spaces.
- foreign
The column name matches the first primary_key of another table
- index
An index will be created
- nullable
When passed, sqlalchemy will get nullable=False
- primary_id
Two modes: with and without an int qualifier. With a qualifier, makes the column a primary_key using the int qualifier as a base for the sequence. The int must be between 0 and 100 (exclusive). The sequence is a series: 10<nn>, 20<nn>, … The advantage to this is that all primary_ids are globally unique and can be “type checked” by higher level code.
When used without a qualifier, refers to the primary_id of another table by exactly the same name. A ForeignKeyConstraint and Index will be added to that other table.
- primary_key
Passes primary_key=True
- unique
Column must have unique values (index will be created)
Table constraints:
- foreign
Iterable of arguments to sqlalchemy.ForeignKeyConstraint. Also, creates an index.
- index
Iterable of arguments to sqlalchemy.Index except for the index name which is always created.
- unique
Iterable of arguments to sqlalchemy.UniqueConstraint.
- Parameters:
- exception pykern.sql_db.MoreThanOneRow(**context)[source]¶
Bases:
BaseExcExpected exactly one row, but got more than one.
pykern.util module¶
Support routines, including run dir resolution.
- copyright:
Copyright (c) 2023 RadiaSoft LLC. All Rights Reserved.
- license:
- exception pykern.util.APIError(fmt, *args, **kwargs)[source]¶
Bases:
ExceptionApplication level errors or exceptions sent to client and raised on client
- pykern.util.cfg_absolute_dir(value)[source]¶
Parser function for absolute dir config value
- Parameters:
value (str) – string absolute path to dir
- Returns:
py.path.Local absolute path to dir
- pykern.util.dev_run_dir(package_object)[source]¶
Resolves run directory if in development mode and creates it if it doesn’t exist.
Note that this assumes a file structure like pykern’s where <project_name>’s __init__.py lives in <project_name>/<project_name>/
- Parameters:
package_object (object) – A python object whose root package we are resolving the run directory for
- Returns:
The absolute path to the run directory.
- Return type:
py.path.local
- Raises:
AssertionError – Raised when not in development mode
- pykern.util.is_pure_text(value, is_truncated=False)[source]¶
Guesses if value is text data using heuristics:
Checks if value can be utf-8 decoded.
If fails to decode and is_truncated, probes backwards up to 4 chars (4 is maximum length of a utf-8 char) in case a valid utf-8 char was truncated at the boundary of value.
Returns False if null byte is present.
On successful decode, checks that the amount of control codes not typical of text data do not exceed one third of the total characters.
- pykern.util.random_base62(length=16)[source]¶
Returns a safe string of sufficient length to be a nonce.
The default is 62^16, which is 4e28. For comparison, 2^64 is 1e19 and 2^128 is 3e38.
- pykern.util.unbound_localhost_inet_port(protocol, start=10000, stop=20000)[source]¶
Looks for AF_INET protocol port for which bind succeeds
pykern.xlsx module¶
Excel spreadsheet generator
- copyright:
Copyright (c) 2021 RadiaSoft LLC. All Rights Reserved.
- license: