in statefun-sdk-python/statefun/core.py [0:0]
def __init__(self,
name: str,
type: Type,
expire_after_call: timedelta = None,
expire_after_write: timedelta = None):
"""
ValueSpec a specification of a persisted value.
:param name: a user defined state name, used to represent this value. A name must be lower case, alphanumeric string
without spaces, that starts with either a letter or an underscore (_)
:param type: the value's Type. (see
statefun.Type) :param expire_after_call: expire (remove) this value if a call to this function hasn't been
made for the given duration.
:param expire_after_write: expire this value if it wasn't written to for the given duration.
"""
if not name:
raise ValueError("name can not be missing.")
if not name.isidentifier():
raise ValueError(
f"invalid name {name}. A spec name can only contains alphanumeric letters (a-z) and (0-9), "
f"or underscores ( "
f"_). A valid identifier cannot start with a number, or contain any spaces.")
if iskeyword(name):
forbidden = '\n'.join(kwlist)
raise ValueError(
f"invalid spec name {name} (Python SDK specifically). since {name} will result as an attribute on "
f"context.store.\n"
f"The following names are forbidden:\n {forbidden}")
if not name.islower():
raise ValueError(f"Only lower case names are allowed, {name} is given.")
self.name = name
if not type:
raise ValueError("type can not be missing.")
if not isinstance(type, Type):
raise TypeError("type is not a StateFun type.")
self.type = type
if expire_after_call and expire_after_write:
# both can not be set.
raise ValueError("Either expire_after_call or expire_after_write can be set, but not both.")
if expire_after_call:
self.duration = int(expire_after_call.total_seconds() * 1000.0)
self.after_call = True
self.after_write = False
elif expire_after_write:
self.duration = int(expire_after_write.total_seconds() * 1000.0)
self.after_call = False
self.after_write = True
else:
self.duration = 0
self.after_call = False
self.after_write = False