Skip to content

Common

Shared kernel used by other domains. Narrative guide: Common. Tracing how-to: Tracing.

Config

pypepper.common.config

YAML config loading and typed configuration models.

config module-attribute 

config = Config()

ConfHTTPServer

Source code in pypepper/common/config.py 
13
14
15
class ConfHTTPServer:
    enable: bool
    port: int

enable instance-attribute 

enable: bool

port instance-attribute 

port: int

ConfHTTPSServer

Source code in pypepper/common/config.py 
18
19
20
21
22
23
24
class ConfHTTPSServer:
    enable: bool
    port: int
    mutualTLS: bool
    certFile: str = ""
    keyFile: str = ""
    caFile: str = ""

enable instance-attribute 

enable: bool

port instance-attribute 

port: int

mutualTLS instance-attribute 

mutualTLS: bool

certFile class-attribute instance-attribute 

certFile: str = ''

keyFile class-attribute instance-attribute 

keyFile: str = ''

caFile class-attribute instance-attribute 

caFile: str = ''

ConfNetwork

Source code in pypepper/common/config.py 
27
28
29
30
class ConfNetwork:
    ip: str
    httpServer: ConfHTTPServer
    httpsServer: ConfHTTPSServer

ip instance-attribute 

ip: str

httpServer instance-attribute 

httpServer: ConfHTTPServer

httpsServer instance-attribute 

httpsServer: ConfHTTPSServer

ConfLog

Source code in pypepper/common/config.py 
33
34
35
class ConfLog:
    level: str
    colorize: bool

level instance-attribute 

level: str

colorize instance-attribute 

colorize: bool

ConfSSEAuthentication

Source code in pypepper/common/config.py 
38
39
40
class ConfSSEAuthentication:
    enabled: bool
    validKeys: list

enabled instance-attribute 

enabled: bool

validKeys instance-attribute 

validKeys: list

ConfSSERateLimit

Source code in pypepper/common/config.py 
43
44
45
class ConfSSERateLimit:
    enabled: bool
    maxRequestsPerMinute: int

enabled instance-attribute 

enabled: bool

maxRequestsPerMinute instance-attribute 

maxRequestsPerMinute: int

ConfSSE

Source code in pypepper/common/config.py 
48
49
50
51
52
53
54
class ConfSSE:
    maxTotalConnections: int
    maxConnectionsPerIP: int
    maxQueueSize: int
    streamTimeoutSeconds: int
    authentication: ConfSSEAuthentication
    rateLimit: ConfSSERateLimit

maxTotalConnections instance-attribute 

maxTotalConnections: int

maxConnectionsPerIP instance-attribute 

maxConnectionsPerIP: int

maxQueueSize instance-attribute 

maxQueueSize: int

streamTimeoutSeconds instance-attribute 

streamTimeoutSeconds: int

authentication instance-attribute 

authentication: ConfSSEAuthentication

rateLimit instance-attribute 

rateLimit: ConfSSERateLimit

ConfTracingOTLP

Source code in pypepper/common/config.py 
57
58
59
class ConfTracingOTLP:
    enabled: bool
    endpoint: str

enabled instance-attribute 

enabled: bool

endpoint instance-attribute 

endpoint: str

ConfTracing

Source code in pypepper/common/config.py 
62
63
64
65
66
class ConfTracing:
    enabled: bool
    serviceName: str
    console: bool
    otlp: ConfTracingOTLP

enabled instance-attribute 

enabled: bool

serviceName instance-attribute 

serviceName: str

console instance-attribute 

console: bool

otlp instance-attribute 

otlp: ConfTracingOTLP

ConfSchedulerJobStore

Source code in pypepper/common/config.py 
69
70
71
72
73
74
75
76
77
78
79
class ConfSchedulerJobStore:
    backend: str
    uri: str | None
    host: str | None
    port: int | None
    username: str | None
    password: str | None
    db: str | None
    sslmode: str | None
    charset: str | None
    auth_source: str | None

backend instance-attribute 

backend: str

uri instance-attribute 

uri: str | None

host instance-attribute 

host: str | None

port instance-attribute 

port: int | None

username instance-attribute 

username: str | None

password instance-attribute 

password: str | None

db instance-attribute 

db: str | None

sslmode instance-attribute 

sslmode: str | None

charset instance-attribute 

charset: str | None

auth_source instance-attribute 

auth_source: str | None

ConfScheduler

Source code in pypepper/common/config.py 
82
83
class ConfScheduler:
    jobStore: ConfSchedulerJobStore

jobStore instance-attribute 

jobStore: ConfSchedulerJobStore

YmlConfig

Source code in pypepper/common/config.py 
86
87
88
89
90
91
92
class YmlConfig:
    network: ConfNetwork
    log: ConfLog
    sse: ConfSSE
    tracing: ConfTracing
    scheduler: ConfScheduler
    custom: Any

network instance-attribute 

network: ConfNetwork

log instance-attribute 

log: ConfLog

sse instance-attribute 

sse: ConfSSE

tracing instance-attribute 

tracing: ConfTracing

scheduler instance-attribute 

scheduler: ConfScheduler

custom instance-attribute 

custom: Any

Config

Source code in pypepper/common/config.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class Config:
    _default_config_path = "./conf/"
    _default_config_filename = "app.config.yaml"
    _default_config_filepath = os.path.join(_default_config_path, _default_config_filename)

    def __init__(self) -> None:
        self._setting: Any = None
        self._deferred_durable_job_store_backend: str | None = None

    def _get_parser(self, **parser_kwargs):
        parser = argparse.ArgumentParser(**parser_kwargs)
        parser.add_argument(
            "-c",
            "--config",
            type=str,
            const=True,
            default=os.path.join(self._default_config_filepath),
            nargs="?",
            help="config filename & path",
        )
        return parser

    def load_config(self, filename: str | None = None):
        if filename:
            service_config_filename = os.path.abspath(filename)
        else:
            parser = self._get_parser()
            args = parser.parse_args()
            service_config_filename = args.config or os.path.abspath(self._default_config_filepath)

        with open(service_config_filename) as fd:
            data = fd.read()
        self._setting = Box(yaml.safe_load(data))

        # Set log config (level, colorize...)
        if hasattr(self.get_yml_config(), "log") and hasattr(self.get_yml_config().log, "level"):
            log.set_log_level(self.get_yml_config().log.level)
            log.set_colorize(self.get_yml_config().log.colorize)

        from pypepper.common.tracing import setup_from_config

        setup_from_config(self.get_yml_config())
        self.refresh_scheduler_job_store_deferred()

    def refresh_scheduler_job_store_deferred(self) -> None:
        """
        Re-read durable ``scheduler.jobStore`` from the current YAML into the deferred flag.

        Counterpart to :meth:`mark_scheduler_job_store_applied` (used by ``reset_job_store``
        and ``load_config``). Memory / missing backends clear the flag.
        """
        self._deferred_durable_job_store_backend = None
        yml = self.get_yml_config()
        if yml is None or not hasattr(yml, "scheduler") or yml.scheduler is None:
            return
        job_store = getattr(yml.scheduler, "jobStore", None)
        if job_store is None:
            return
        backend = getattr(job_store, "backend", None)
        if backend is None:
            return
        name = str(backend).strip().lower()
        if name in ("", "memory"):
            return
        self._deferred_durable_job_store_backend = str(backend)

    def mark_scheduler_job_store_applied(self) -> None:
        """Clear the deferred durable jobStore flag (called after setup/configure)."""
        self._deferred_durable_job_store_backend = None

    def ensure_scheduler_job_store_applied(self, *, using_default_memory_store: bool = True) -> None:
        """
        Raise if YAML declared a durable jobStore that has not been applied yet.

        When a non-memory store is already installed (``using_default_memory_store=False``),
        treat the deferred declaration as satisfied so configure-before-load and reload
        after setup do not false-positive.
        """
        backend = self._deferred_durable_job_store_backend
        if backend is None:
            return
        if not using_default_memory_store:
            self.mark_scheduler_job_store_applied()
            return
        raise ValueError(
            f"scheduler.jobStore.backend={backend!r} is present in YAML but not applied by "
            "load_config; call pypepper.scheduler.store.setup_from_config(...) after load "
            "before persisting jobs"
        )

    def get_yml_config(self) -> YmlConfig:
        return self._setting

load_config 

load_config(filename: str | None = None)
Source code in pypepper/common/config.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def load_config(self, filename: str | None = None):
    if filename:
        service_config_filename = os.path.abspath(filename)
    else:
        parser = self._get_parser()
        args = parser.parse_args()
        service_config_filename = args.config or os.path.abspath(self._default_config_filepath)

    with open(service_config_filename) as fd:
        data = fd.read()
    self._setting = Box(yaml.safe_load(data))

    # Set log config (level, colorize...)
    if hasattr(self.get_yml_config(), "log") and hasattr(self.get_yml_config().log, "level"):
        log.set_log_level(self.get_yml_config().log.level)
        log.set_colorize(self.get_yml_config().log.colorize)

    from pypepper.common.tracing import setup_from_config

    setup_from_config(self.get_yml_config())
    self.refresh_scheduler_job_store_deferred()

refresh_scheduler_job_store_deferred 

refresh_scheduler_job_store_deferred() -> None

Re-read durable scheduler.jobStore from the current YAML into the deferred flag.

Counterpart to :meth:mark_scheduler_job_store_applied (used by reset_job_store and load_config). Memory / missing backends clear the flag.

Source code in pypepper/common/config.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def refresh_scheduler_job_store_deferred(self) -> None:
    """
    Re-read durable ``scheduler.jobStore`` from the current YAML into the deferred flag.

    Counterpart to :meth:`mark_scheduler_job_store_applied` (used by ``reset_job_store``
    and ``load_config``). Memory / missing backends clear the flag.
    """
    self._deferred_durable_job_store_backend = None
    yml = self.get_yml_config()
    if yml is None or not hasattr(yml, "scheduler") or yml.scheduler is None:
        return
    job_store = getattr(yml.scheduler, "jobStore", None)
    if job_store is None:
        return
    backend = getattr(job_store, "backend", None)
    if backend is None:
        return
    name = str(backend).strip().lower()
    if name in ("", "memory"):
        return
    self._deferred_durable_job_store_backend = str(backend)

mark_scheduler_job_store_applied 

mark_scheduler_job_store_applied() -> None

Clear the deferred durable jobStore flag (called after setup/configure).

Source code in pypepper/common/config.py 
161
162
163
def mark_scheduler_job_store_applied(self) -> None:
    """Clear the deferred durable jobStore flag (called after setup/configure)."""
    self._deferred_durable_job_store_backend = None

ensure_scheduler_job_store_applied 

ensure_scheduler_job_store_applied(
    *, using_default_memory_store: bool = True
) -> None

Raise if YAML declared a durable jobStore that has not been applied yet.

When a non-memory store is already installed (using_default_memory_store=False), treat the deferred declaration as satisfied so configure-before-load and reload after setup do not false-positive.

Source code in pypepper/common/config.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def ensure_scheduler_job_store_applied(self, *, using_default_memory_store: bool = True) -> None:
    """
    Raise if YAML declared a durable jobStore that has not been applied yet.

    When a non-memory store is already installed (``using_default_memory_store=False``),
    treat the deferred declaration as satisfied so configure-before-load and reload
    after setup do not false-positive.
    """
    backend = self._deferred_durable_job_store_backend
    if backend is None:
        return
    if not using_default_memory_store:
        self.mark_scheduler_job_store_applied()
        return
    raise ValueError(
        f"scheduler.jobStore.backend={backend!r} is present in YAML but not applied by "
        "load_config; call pypepper.scheduler.store.setup_from_config(...) after load "
        "before persisting jobs"
    )

get_yml_config 

get_yml_config() -> YmlConfig
Source code in pypepper/common/config.py 
185
186
def get_yml_config(self) -> YmlConfig:
    return self._setting

Log

pypepper.common.log

Loguru-based process logger helpers.

default_log_filter module-attribute 

default_log_filter = LogLevelFilter(LogLevel.TRACE)

log_fmt module-attribute 

log_fmt = "[<green>{time:YYYY-MM-DDTHH:mm:ss.SSSZ}</green>][<level>{level:<8}</level>][<cyan>$host:$user</cyan>][<cyan>pid:{process}|tid:{thread}</cyan>][<cyan>{file.path}:{line}</cyan>][<cyan>{module}.{function}</cyan>][<magenta>{extra[req_id]}</magenta>][<level>{message}</level>]"

log_format_template module-attribute 

log_format_template = Template(log_fmt).substitute(
    host=socket.gethostname(), user=getpass.getuser()
)

config module-attribute 

config = {
    "handlers": [
        {
            "sink": sys.stdout,
            "format": log_format_template,
            "level": LogLevel.TRACE,
            "colorize": True,
            "enqueue": True,
            "filter": default_log_filter,
        }
    ],
    "levels": [
        dict(
            name="FATAL",
            no=60,
            icon="☠",
            color="<RED><bold>",
        )
    ],
    "extra": {"req_id": 0},
}

log module-attribute 

log = Logger()

LogLevel

Bases: str, Enum

Log level

Source code in pypepper/common/log.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class LogLevel(str, Enum):
    """
    Log level
    """

    TRACE = "TRACE"
    DEBUG = "DEBUG"
    INFO = "INFO"
    WARNING = "WARNING"
    ERROR = "ERROR"
    FATAL = "FATAL"

    @classmethod
    def has_name(cls, name) -> bool:
        """
        Check if the name is in LogLevel
        :param name: log level name
        :return: result
        """

        return name in cls.__members__

    @classmethod
    def has_value(cls, value) -> bool:
        """
        Check if the value is in LogLevel
        :param value: log level value
        :return: result
        """

        return value in cls.__members__.values()

TRACE class-attribute instance-attribute 

TRACE = 'TRACE'

DEBUG class-attribute instance-attribute 

DEBUG = 'DEBUG'

INFO class-attribute instance-attribute 

INFO = 'INFO'

WARNING class-attribute instance-attribute 

WARNING = 'WARNING'

ERROR class-attribute instance-attribute 

ERROR = 'ERROR'

FATAL class-attribute instance-attribute 

FATAL = 'FATAL'

has_name classmethod 

has_name(name) -> bool

Check if the name is in LogLevel

Parameters:

Name Type Description Default
name

log level name

 required

Returns:

Type Description
bool

result
Source code in pypepper/common/log.py 
28
29
30
31
32
33
34
35
36
@classmethod
def has_name(cls, name) -> bool:
    """
    Check if the name is in LogLevel
    :param name: log level name
    :return: result
    """

    return name in cls.__members__

has_value classmethod 

has_value(value) -> bool

Check if the value is in LogLevel

Parameters:

Name Type Description Default
value

log level value

 required

Returns:

Type Description
bool

result
Source code in pypepper/common/log.py 
38
39
40
41
42
43
44
45
46
@classmethod
def has_value(cls, value) -> bool:
    """
    Check if the value is in LogLevel
    :param value: log level value
    :return: result
    """

    return value in cls.__members__.values()

LogLevelFilter

Log level filter

Source code in pypepper/common/log.py 
49
50
51
52
53
54
55
56
57
58
59
class LogLevelFilter:
    """
    Log level filter
    """

    def __init__(self, level):
        self.level = level

    def __call__(self, record):
        level_no = logger.level(self.level).no
        return record["level"].no >= level_no

level instance-attribute 

level = level

Logger

Source code in pypepper/common/log.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class Logger:
    def __init__(self):
        self._logger = logger.opt(depth=1)

    def get_logger(self):
        return self._logger

    def request_id(self, req_id=0):
        """
        Return a temporary logger bound with request id.
        Does not mutate the process-wide logger singleton.
        """
        bound = Logger()
        bound._logger = self._logger.bind(req_id=req_id)
        return bound

    def logo(self, msg: str):
        msg += json.dumps(version.get_version_info(), indent=4)
        self._logger.info(msg)

    # Severity: 5
    def trace(self, msg: str, *args: Any, **kwargs: Any):
        self._logger.trace(msg, *args, **kwargs)

    # Severity: 10
    def debug(self, msg: str, *args: Any, **kwargs: Any):
        self._logger.debug(msg, *args, **kwargs)

    # Severity: 20
    def info(self, msg: str, *args: Any, **kwargs: Any):
        self._logger.info(msg, *args, **kwargs)

    # Severity: 30
    def warn(self, msg: str, *args: Any, **kwargs: Any):
        self._logger.warning(msg, *args, **kwargs)

    # Severity: 40
    def error(self, msg: str, *args: Any, **kwargs: Any):
        self._logger.error(msg, *args, **kwargs)

    # Severity: 60
    def fatal(self, msg: str, *args: Any, **kwargs: Any):
        self._logger.log(LogLevel.FATAL, msg, *args, **kwargs)

    def close(self):
        self._logger.complete()
        self._logger.remove()

    @staticmethod
    def set_log_level(level: str):
        if LogLevel.has_value(level):
            default_log_filter.level = level

    @staticmethod
    def set_colorize(colorize=True):
        config["handlers"][0]["colorize"] = colorize  # type: ignore[index]
        logger.remove()
        logger.configure(handlers=config["handlers"])  # type: ignore[arg-type]

get_logger 

get_logger()
Source code in pypepper/common/log.py 
110
111
def get_logger(self):
    return self._logger

request_id 

request_id(req_id=0)

Return a temporary logger bound with request id. Does not mutate the process-wide logger singleton.

Source code in pypepper/common/log.py 
113
114
115
116
117
118
119
120
def request_id(self, req_id=0):
    """
    Return a temporary logger bound with request id.
    Does not mutate the process-wide logger singleton.
    """
    bound = Logger()
    bound._logger = self._logger.bind(req_id=req_id)
    return bound

logo(msg: str)
Source code in pypepper/common/log.py 
122
123
124
def logo(self, msg: str):
    msg += json.dumps(version.get_version_info(), indent=4)
    self._logger.info(msg)

trace 

trace(msg: str, *args: Any, **kwargs: Any)
Source code in pypepper/common/log.py 
127
128
def trace(self, msg: str, *args: Any, **kwargs: Any):
    self._logger.trace(msg, *args, **kwargs)

debug 

debug(msg: str, *args: Any, **kwargs: Any)
Source code in pypepper/common/log.py 
131
132
def debug(self, msg: str, *args: Any, **kwargs: Any):
    self._logger.debug(msg, *args, **kwargs)

info 

info(msg: str, *args: Any, **kwargs: Any)
Source code in pypepper/common/log.py 
135
136
def info(self, msg: str, *args: Any, **kwargs: Any):
    self._logger.info(msg, *args, **kwargs)

warn 

warn(msg: str, *args: Any, **kwargs: Any)
Source code in pypepper/common/log.py 
139
140
def warn(self, msg: str, *args: Any, **kwargs: Any):
    self._logger.warning(msg, *args, **kwargs)

error 

error(msg: str, *args: Any, **kwargs: Any)
Source code in pypepper/common/log.py 
143
144
def error(self, msg: str, *args: Any, **kwargs: Any):
    self._logger.error(msg, *args, **kwargs)

fatal 

fatal(msg: str, *args: Any, **kwargs: Any)
Source code in pypepper/common/log.py 
147
148
def fatal(self, msg: str, *args: Any, **kwargs: Any):
    self._logger.log(LogLevel.FATAL, msg, *args, **kwargs)

close 

close()
Source code in pypepper/common/log.py 
150
151
152
def close(self):
    self._logger.complete()
    self._logger.remove()

set_log_level staticmethod 

set_log_level(level: str)
Source code in pypepper/common/log.py 
154
155
156
157
@staticmethod
def set_log_level(level: str):
    if LogLevel.has_value(level):
        default_log_filter.level = level

set_colorize staticmethod 

set_colorize(colorize=True)
Source code in pypepper/common/log.py 
159
160
161
162
163
@staticmethod
def set_colorize(colorize=True):
    config["handlers"][0]["colorize"] = colorize  # type: ignore[index]
    logger.remove()
    logger.configure(handlers=config["handlers"])  # type: ignore[arg-type]

Options

pypepper.common.options

Functional options pattern helpers.

IOptions

Options interface

Source code in pypepper/common/options.py 
13
14
15
16
17
18
19
20
21
22
23
class IOptions(metaclass=ABCMeta):
    """
    Options interface
    """

    dryrun: bool = False
    context: IContext

    def __init__(self, context: IContext, dryrun: bool = False):
        self.context = context
        self.dryrun = dryrun

context instance-attribute 

context: IContext = context

dryrun class-attribute instance-attribute 

dryrun: bool = dryrun

new 

new(
    option_funcs: tuple[Callable[[IOptions], Any], ...]
    | None = (),
) -> IOptions

New options.

Parameters:

Name Type Description Default
option_funcs tuple[Callable[[IOptions], Any], ...] | None

option functions.

 ()

Returns:

Type Description
IOptions

options.
Source code in pypepper/common/options.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def new(option_funcs: tuple[Callable[[IOptions], Any], ...] | None = ()) -> IOptions:
    """
    New options.
    :param option_funcs: option functions.
    :return: options.
    """

    opts = IOptions(
        dryrun=False,
        context=new_context(),
    )

    for func in option_funcs or ():
        func(opts)

    return opts

with_context 

with_context(ctx: IContext) -> Callable[[IOptions], Any]

With context options

Parameters:

Name Type Description Default
ctx IContext

context

 required

Returns:

Type Description
Callable[[IOptions], Any]

init context function
Source code in pypepper/common/options.py 
44
45
46
47
48
49
50
51
52
53
54
def with_context(ctx: IContext) -> Callable[[IOptions], Any]:
    """
    With context options
    :param ctx: context
    :return: init context function
    """

    def f(opts: IOptions):
        opts.context = ctx

    return f

with_dryrun 

with_dryrun(is_dryrun: bool) -> Callable[[IOptions], Any]

With dryrun options

Parameters:

Name Type Description Default
is_dryrun bool

dryrun True/False

 required

Returns:

Type Description
Callable[[IOptions], Any]

init dryrun function
Source code in pypepper/common/options.py 
57
58
59
60
61
62
63
64
65
66
67
def with_dryrun(is_dryrun: bool) -> Callable[[IOptions], Any]:
    """
    With dryrun options
    :param is_dryrun: dryrun True/False
    :return: init dryrun function
    """

    def f(opts: IOptions):
        opts.dryrun = is_dryrun

    return f

Context

pypepper.common.context

Chained request/job context exports.

Context

Bases: IContext

Context

Source code in pypepper/common/context/context.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class Context(IContext):
    """
    Context
    """

    def __init__(
        self,
        context_id: str | None = None,
        parent: IContext | None = None,
    ):
        self.index = 0
        self.context_id = context_id
        self.context = {}
        self.parent = parent

        if parent:
            self.index = parent.index + 1

    def with_value(self, key: Any, value: Any) -> IContext:
        """
        Set context key/value
        :param key: key
        :param value: value
        :return: Context
        """

        self.context[key] = value
        return self

    def trace(self, index: int) -> IContext:
        """
        Trace to the context by index.
        :param index: context index.
        :return: Context.
        :raises InternalException: if the index is not present in the chain
        """

        if self.index == index:
            return self

        if self.parent is None:
            raise InternalException(ERROR_CONTEXT_INDEX_NOT_FOUND)

        return self.parent.trace(index)

    def length(self) -> int:
        """
        Get the length from the context with index 0 to the current context
        :return: length
        """

        return self.index + 1

    def head(self) -> IContext:
        """
        Get context chain head
        :return: context chain head
        """

        return self.trace(0)

index instance-attribute 

index = 0

context_id instance-attribute 

context_id = context_id

context instance-attribute 

context = {}

parent instance-attribute 

parent = parent

with_value 

with_value(key: Any, value: Any) -> IContext

Set context key/value

Parameters:

Name Type Description Default
key Any

key

 required
value Any

value

 required

Returns:

Type Description
IContext

Context
Source code in pypepper/common/context/context.py 
35
36
37
38
39
40
41
42
43
44
def with_value(self, key: Any, value: Any) -> IContext:
    """
    Set context key/value
    :param key: key
    :param value: value
    :return: Context
    """

    self.context[key] = value
    return self

trace 

trace(index: int) -> IContext

Trace to the context by index.

Parameters:

Name Type Description Default
index int

context index.

 required

Returns:

Type Description
IContext

Context.

Raises:

Type Description
InternalException

if the index is not present in the chain
Source code in pypepper/common/context/context.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def trace(self, index: int) -> IContext:
    """
    Trace to the context by index.
    :param index: context index.
    :return: Context.
    :raises InternalException: if the index is not present in the chain
    """

    if self.index == index:
        return self

    if self.parent is None:
        raise InternalException(ERROR_CONTEXT_INDEX_NOT_FOUND)

    return self.parent.trace(index)

length 

length() -> int

Get the length from the context with index 0 to the current context :return: length

Source code in pypepper/common/context/context.py 
62
63
64
65
66
67
68
def length(self) -> int:
    """
    Get the length from the context with index 0 to the current context
    :return: length
    """

    return self.index + 1

head 

head() -> IContext

Get context chain head :return: context chain head

Source code in pypepper/common/context/context.py 
70
71
72
73
74
75
76
def head(self) -> IContext:
    """
    Get context chain head
    :return: context chain head
    """

    return self.trace(0)

born 

born(
    length: int,
    parent: Context | None = None,
    id_provider: Callable[P, T] | None = None,
) -> Context

Born the context chain.

Parameters:

Name Type Description Default
length int

chain length.

 required
parent Context | None

parent context.

 None
id_provider Callable[P, T] | None

context ID provider.

 None

Returns:

Type Description
Context

context.
Source code in pypepper/common/context/context.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def born(
    length: int,
    parent: Context | None = None,
    id_provider: Callable[P, T] | None = None,
) -> Context:
    """
    Born the context chain.
    :param length: chain length.
    :param parent: parent context.
    :param id_provider: context ID provider.
    :return: context.
    """

    if parent:
        if parent.index == length - 1:
            return parent

        next_id = id_provider() if id_provider is not None else uuid.new_uuid()  # type: ignore[call-arg]
        child = new(parent=parent, context_id=str(next_id))

        return born(length=length, parent=child, id_provider=id_provider)

    next_id = id_provider() if id_provider is not None else uuid.new_uuid()  # type: ignore[call-arg]
    head = new(context_id=str(next_id))

    return born(
        length=length,
        parent=head,
        id_provider=id_provider,
    )

new 

new(
    context_id: str | None = None,
    parent: Context | None = None,
) -> Context

New context.

Parameters:

Name Type Description Default
context_id str | None

new context ID (optional).

 None
parent Context | None

the parent context (optional).

 None

Returns:

Type Description
Context

context.
Source code in pypepper/common/context/context.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def new(
    context_id: str | None = None,
    parent: Context | None = None,
) -> Context:
    """
    New context.
    :param context_id: new context ID (optional).
    :param parent: the parent context (optional).
    :return: context.
    """

    return Context(
        context_id=context_id,
        parent=parent,
    )

Cache

pypepper.common.cache

Thread-safe TTL cache set.

Cache

Bases: TTLCache

Thread safe TTL cache

Source code in pypepper/common/cache.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class Cache(TTLCache):
    """
    Thread safe TTL cache
    """

    default_cache_maxsize = 128
    default_cache_ttl = 60

    def __init__(
        self,
        maxsize: int = default_cache_maxsize,
        ttl: float = default_cache_ttl,
    ):
        super().__init__(maxsize, ttl)
        self._lock = Lock()

    def set(self, key: Any, value: Any) -> None:
        """
        Set key/value
        :param key: cache key
        :param value: cache value
        :return: None
        """

        with self._lock:
            try:
                self[key] = value
            except ValueError:
                return None

    def get(self, key: Any, default: Any = None) -> Any | None:
        """
        Get value
        :param key: cache key
        :param default: default value when missing
        :return: cache value
        """

        with self._lock:
            try:
                return self[key] if key else default
            except KeyError:
                return default

default_cache_maxsize class-attribute instance-attribute 

default_cache_maxsize = 128

default_cache_ttl class-attribute instance-attribute 

default_cache_ttl = 60

set 

set(key: Any, value: Any) -> None

Set key/value

Parameters:

Name Type Description Default
key Any

cache key

 required
value Any

cache value

 required

Returns:

Type Description
None

None
Source code in pypepper/common/cache.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
def set(self, key: Any, value: Any) -> None:
    """
    Set key/value
    :param key: cache key
    :param value: cache value
    :return: None
    """

    with self._lock:
        try:
            self[key] = value
        except ValueError:
            return None

get 

get(key: Any, default: Any = None) -> Any | None

Get value

Parameters:

Name Type Description Default
key Any

cache key

 required
default Any

default value when missing

 None

Returns:

Type Description
Any | None

cache value
Source code in pypepper/common/cache.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
def get(self, key: Any, default: Any = None) -> Any | None:
    """
    Get value
    :param key: cache key
    :param default: default value when missing
    :return: cache value
    """

    with self._lock:
        try:
            return self[key] if key else default
        except KeyError:
            return default

CacheSet

A thread safe TTL cache-set

Source code in pypepper/common/cache.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class CacheSet:
    """
    A thread safe TTL cache-set
    """

    def __init__(self):
        self._lock = Lock()
        self._cache_store: MutableMapping[str, Cache] = {}

    def new(
        self,
        name: str,
        maxsize: int = Cache.default_cache_maxsize,
        ttl: float = Cache.default_cache_ttl,
    ) -> Cache:
        """
        Return a named cache, creating it on first use.

        ``maxsize`` / ``ttl`` apply only when the name is created. If the name
        already exists, the existing cache is returned and those params are ignored.
        """

        with self._lock:
            existing = self._cache_store.get(name)
            if existing is None:
                existing = Cache(maxsize, ttl)
                self._cache_store[name] = existing
            return existing

    def get(self, name: str) -> Cache | None:
        """
        Get the cache from cache-set
        :param name: cache name
        :return: cache
        """

        with self._lock:
            return self._cache_store.get(name)

    def clear(self):
        """
        Clear cache-set
        :return: None
        """

        with self._lock:
            for name in self._cache_store:
                self._cache_store[name].clear()
            self._cache_store.clear()

new 

new(
    name: str,
    maxsize: int = Cache.default_cache_maxsize,
    ttl: float = Cache.default_cache_ttl,
) -> Cache

Return a named cache, creating it on first use.

maxsize / ttl apply only when the name is created. If the name already exists, the existing cache is returned and those params are ignored.

Source code in pypepper/common/cache.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def new(
    self,
    name: str,
    maxsize: int = Cache.default_cache_maxsize,
    ttl: float = Cache.default_cache_ttl,
) -> Cache:
    """
    Return a named cache, creating it on first use.

    ``maxsize`` / ``ttl`` apply only when the name is created. If the name
    already exists, the existing cache is returned and those params are ignored.
    """

    with self._lock:
        existing = self._cache_store.get(name)
        if existing is None:
            existing = Cache(maxsize, ttl)
            self._cache_store[name] = existing
        return existing

get 

get(name: str) -> Cache | None

Get the cache from cache-set

Parameters:

Name Type Description Default
name str

cache name

 required

Returns:

Type Description
Cache | None

cache
Source code in pypepper/common/cache.py 
86
87
88
89
90
91
92
93
94
def get(self, name: str) -> Cache | None:
    """
    Get the cache from cache-set
    :param name: cache name
    :return: cache
    """

    with self._lock:
        return self._cache_store.get(name)

clear 

clear()

Clear cache-set :return: None

Source code in pypepper/common/cache.py 
 96
 97
 98
 99
100
101
102
103
104
105
def clear(self):
    """
    Clear cache-set
    :return: None
    """

    with self._lock:
        for name in self._cache_store:
            self._cache_store[name].clear()
        self._cache_store.clear()

new_cache 

new_cache() -> Cache

New cache :return: cache

Source code in pypepper/common/cache.py 
108
109
110
111
112
113
114
def new_cache() -> Cache:
    """
    New cache
    :return: cache
    """

    return Cache()

new_cache_set 

new_cache_set() -> CacheSet

New cache-set. :return: cache-set.

Source code in pypepper/common/cache.py 
117
118
119
120
121
122
123
def new_cache_set() -> CacheSet:
    """
    New cache-set.
    :return: cache-set.
    """

    return CacheSet()

Tracing

pypepper.common.tracing

OpenTelemetry tracing setup and HTTP middleware exports.

TracingMiddleware

Bases: BaseHTTPMiddleware

Create a SERVER span per HTTP request; attach request_id when present.

Source code in pypepper/common/tracing/middleware.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class TracingMiddleware(BaseHTTPMiddleware):
    """Create a SERVER span per HTTP request; attach request_id when present."""

    def __init__(self, app: ASGIApp, tracer_name: str = "pypepper.network.http"):
        super().__init__(app)
        self._tracer_name = tracer_name

    async def dispatch(self, request: Request, call_next) -> Response:
        tracer = get_tracer(self._tracer_name)
        span_name = f"{request.method} {request.url.path}"
        with tracer.start_as_current_span(span_name, kind=SpanKind.SERVER) as span:
            span.set_attribute("http.method", request.method)
            span.set_attribute("http.scheme", request.url.scheme)
            span.set_attribute("url.path", request.url.path)
            if request.url.query:
                span.set_attribute("url.query", request.url.query)

            try:
                response = await call_next(request)
            except Exception as exc:
                span.set_status(Status(StatusCode.ERROR, str(exc)))
                span.record_exception(exc)
                raise

            req_id = getattr(request.state, "request_id", None)
            if req_id is not None:
                span.set_attribute("request_id", str(req_id))

            span.set_attribute("http.status_code", response.status_code)
            if response.status_code >= 500:
                span.set_status(Status(StatusCode.ERROR))
            return response

dispatch async 

dispatch(request: Request, call_next) -> Response
Source code in pypepper/common/tracing/middleware.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
async def dispatch(self, request: Request, call_next) -> Response:
    tracer = get_tracer(self._tracer_name)
    span_name = f"{request.method} {request.url.path}"
    with tracer.start_as_current_span(span_name, kind=SpanKind.SERVER) as span:
        span.set_attribute("http.method", request.method)
        span.set_attribute("http.scheme", request.url.scheme)
        span.set_attribute("url.path", request.url.path)
        if request.url.query:
            span.set_attribute("url.query", request.url.query)

        try:
            response = await call_next(request)
        except Exception as exc:
            span.set_status(Status(StatusCode.ERROR, str(exc)))
            span.record_exception(exc)
            raise

        req_id = getattr(request.state, "request_id", None)
        if req_id is not None:
            span.set_attribute("request_id", str(req_id))

        span.set_attribute("http.status_code", response.status_code)
        if response.status_code >= 500:
            span.set_status(Status(StatusCode.ERROR))
        return response

get_tracer 

get_tracer(name: str = 'pypepper') -> trace.Tracer

Return a tracer from the global provider (no-op when tracing is disabled).

Source code in pypepper/common/tracing/setup.py 
18
19
20
def get_tracer(name: str = "pypepper") -> trace.Tracer:
    """Return a tracer from the global provider (no-op when tracing is disabled)."""
    return trace.get_tracer(name)

setup_for_tests 

setup_for_tests(
    exporter: Any, service_name: str = "pypepper-test"
) -> TracerProvider

Install a TracerProvider that exports to the given span exporter (tests only).

Uses SimpleSpanProcessor so spans are available immediately after the request.

Source code in pypepper/common/tracing/setup.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def setup_for_tests(exporter: Any, service_name: str = "pypepper-test") -> TracerProvider:
    """
    Install a TracerProvider that exports to the given span exporter (tests only).

    Uses SimpleSpanProcessor so spans are available immediately after the request.
    """
    global _provider
    shutdown()
    resource = Resource.create({"service.name": service_name})
    provider = TracerProvider(resource=resource)
    provider.add_span_processor(SimpleSpanProcessor(exporter))
    _allow_tracer_provider_override()
    trace.set_tracer_provider(provider)
    _provider = provider
    return provider

setup_from_config 

setup_from_config(yml_config: Any | None = None) -> None

Configure OpenTelemetry from YAML config.

Expected shape (all optional; missing section disables tracing)::

tracing:
  enabled: false
  serviceName: pypepper
  console: false
  otlp:
    enabled: false
    endpoint: http://127.0.0.1:4318
Source code in pypepper/common/tracing/setup.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def setup_from_config(yml_config: Any | None = None) -> None:
    """
    Configure OpenTelemetry from YAML config.

    Expected shape (all optional; missing section disables tracing)::

        tracing:
          enabled: false
          serviceName: pypepper
          console: false
          otlp:
            enabled: false
            endpoint: http://127.0.0.1:4318
    """
    global _provider, _atexit_registered

    shutdown()

    if yml_config is None or not hasattr(yml_config, "tracing"):
        return

    tracing_cfg = yml_config.tracing
    if tracing_cfg is None:
        return

    enabled = bool(getattr(tracing_cfg, "enabled", False))
    if not enabled:
        return

    service_name = getattr(tracing_cfg, "serviceName", None)
    if not service_name and hasattr(yml_config, "serviceInfo"):
        service_name = getattr(yml_config.serviceInfo, "serviceName", None)
    if not service_name:
        service_name = "pypepper"

    console = bool(getattr(tracing_cfg, "console", False))
    otlp_cfg = getattr(tracing_cfg, "otlp", None)
    otlp_enabled = bool(getattr(otlp_cfg, "enabled", False)) if otlp_cfg is not None else False
    otlp_endpoint = (getattr(otlp_cfg, "endpoint", None) if otlp_cfg is not None else None) or "http://127.0.0.1:4318"

    if not console and not otlp_enabled:
        return

    resource = Resource.create({"service.name": service_name})
    provider = TracerProvider(resource=resource)

    if console:
        # Simple processor so spans appear promptly in the terminal during local demos.
        provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))

    if otlp_enabled:
        from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

        endpoint = str(otlp_endpoint).rstrip("/")
        if not endpoint.endswith("/v1/traces"):
            endpoint = f"{endpoint}/v1/traces"
        # Simple processor so local Jaeger UI sees spans without waiting for a batch flush.
        exporter = OTLPSpanExporter(endpoint=endpoint)
        provider.add_span_processor(SimpleSpanProcessor(exporter))

    _allow_tracer_provider_override()
    trace.set_tracer_provider(provider)
    _provider = provider

    if not _atexit_registered:
        atexit.register(shutdown)
        _atexit_registered = True

shutdown 

shutdown() -> None

Flush and shut down the configured TracerProvider, if any.

Source code in pypepper/common/tracing/setup.py 
29
30
31
32
33
34
35
36
37
def shutdown() -> None:
    """Flush and shut down the configured TracerProvider, if any."""
    global _provider
    if _provider is not None:
        _provider.force_flush()
        _provider.shutdown()
        _provider = None
    _allow_tracer_provider_override()
    trace.set_tracer_provider(trace.NoOpTracerProvider())

Crypto

pypepper.common.security.crypto.digest

Hash digest helpers (bytes and hex).

BinaryLike module-attribute 

BinaryLike = str | bytes | bytearray | memoryview

get 

get(data: BinaryLike, alg: str) -> bytes

Get hash (bytes)

Parameters:

Name Type Description Default
data BinaryLike

data in BinaryLike style

 required
alg str

algorithm (md5 / sha1 are rejected)

 required

Returns:

Type Description
bytes

hash (bytes)

Raises:

Type Description
ValueError

if alg is md5 or sha1 (case-insensitive)
Source code in pypepper/common/security/crypto/digest.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def get(data: BinaryLike, alg: str) -> bytes:
    """
    Get hash (bytes)

    :param data: data in BinaryLike style
    :param alg: algorithm (``md5`` / ``sha1`` are rejected)
    :return: hash (bytes)
    :raises ValueError: if ``alg`` is ``md5`` or ``sha1`` (case-insensitive)
    """

    name = alg.lower()
    if name in _REJECTED_DIGEST_ALGS:
        raise ValueError(f"digest algorithm {alg!r} is not allowed; use sha256 or stronger")

    h = hashlib.new(alg)

    if isinstance(data, str):
        h.update(bytes(data, "UTF-8"))
    else:
        h.update(data)

    return h.digest()

get_hex_str 

get_hex_str(data: BinaryLike, alg: str) -> str

Get hash string (hex)

Parameters:

Name Type Description Default
data BinaryLike

data in BinaryLike style

 required
alg str

algorithm (md5 / sha1 are rejected)

 required

Returns:

Type Description
str

hash string (hex)

Raises:

Type Description
ValueError

if alg is md5 or sha1 (case-insensitive)
Source code in pypepper/common/security/crypto/digest.py 
36
37
38
39
40
41
42
43
44
45
46
def get_hex_str(data: BinaryLike, alg: str) -> str:
    """
    Get hash string (hex)

    :param data: data in BinaryLike style
    :param alg: algorithm (``md5`` / ``sha1`` are rejected)
    :return: hash string (hex)
    :raises ValueError: if ``alg`` is ``md5`` or ``sha1`` (case-insensitive)
    """

    return get(data, alg).hex()

pypepper.common.security.crypto.salt

Random salt generation.

MIN_SALT_SIZE module-attribute 

MIN_SALT_SIZE = 16

DEFAULT_SALT_SIZE module-attribute 

DEFAULT_SALT_SIZE = 64

new 

new(size: int = DEFAULT_SALT_SIZE) -> bytes

Generates a random salt of the specified size. The salt should be as unique as possible. It is recommended that a salt is random and at least 16 bytes long. See NIST SP 800-132 for details. Ref: https://nvlpubs.nist.gov/nistpubs/Legacy/SP/nistspecialpublication800-132.pdf

Parameters:

Name Type Description Default
size int

number of bytes

 DEFAULT_SALT_SIZE

Returns:

Type Description
bytes

a random salt
Source code in pypepper/common/security/crypto/salt.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
def new(size: int = DEFAULT_SALT_SIZE) -> bytes:
    """
    Generates a random salt of the specified size.
    The salt should be as unique as possible. It is recommended that a salt is random and at least 16 bytes long.
    See NIST SP 800-132 for details. Ref: https://nvlpubs.nist.gov/nistpubs/Legacy/SP/nistspecialpublication800-132.pdf
    :param size: number of bytes
    :return: a random salt
    """

    if size < MIN_SALT_SIZE:
        raise InternalException("salt size at least 16 bytes long")

    return secrets.token_bytes(size)

pypepper.common.security.crypto.elliptic.ecdsa

ECDSA sign and verify on secp256k1.

ecdsa module-attribute 

ecdsa = ECDSA()

ECDSA

Bases: IElliptic

Source code in pypepper/common/security/crypto/elliptic/ecdsa.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class ECDSA(IElliptic):
    def new_key_pair(self) -> ec.EllipticCurvePrivateKey:
        return ec.generate_private_key(ec.SECP256K1())

    def sign(self, data: bytes, certificate: bytes, hash_alg: str, passphrase: bytes | None = None) -> bytes:
        private_key = serialization.load_pem_private_key(certificate, passphrase)
        if not isinstance(private_key, ec.EllipticCurvePrivateKey):
            raise TypeError("certificate must be an elliptic curve private key")

        return private_key.sign(data=data, signature_algorithm=ec.ECDSA(algorithm.get_hash_algorithm(hash_alg)))

    def verify(self, data: bytes, certificate: bytes, sig: bytes, hash_alg: str) -> bool:
        public_key = serialization.load_pem_public_key(certificate)
        if not isinstance(public_key, ec.EllipticCurvePublicKey):
            raise TypeError("certificate must be an elliptic curve public key")

        try:
            public_key.verify(
                signature=sig, data=data, signature_algorithm=ec.ECDSA(algorithm.get_hash_algorithm(hash_alg))
            )
        except exceptions.InvalidSignature:
            return False

        return True

    @staticmethod
    def get_private_key_pem(private_key: ec.EllipticCurvePrivateKey, passphrase: bytes | None = None) -> bytes:
        encryption_alg: KeySerializationEncryption
        if passphrase:
            encryption_alg = serialization.BestAvailableEncryption(passphrase)
        else:
            encryption_alg = serialization.NoEncryption()
        return private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=encryption_alg,
        )

    @staticmethod
    def get_public_key_pem(private_key: ec.EllipticCurvePrivateKey) -> bytes:
        return private_key.public_key().public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo,
        )

new_key_pair 

new_key_pair() -> ec.EllipticCurvePrivateKey
Source code in pypepper/common/security/crypto/elliptic/ecdsa.py 
13
14
def new_key_pair(self) -> ec.EllipticCurvePrivateKey:
    return ec.generate_private_key(ec.SECP256K1())

sign 

sign(
    data: bytes,
    certificate: bytes,
    hash_alg: str,
    passphrase: bytes | None = None,
) -> bytes
Source code in pypepper/common/security/crypto/elliptic/ecdsa.py 
16
17
18
19
20
21
def sign(self, data: bytes, certificate: bytes, hash_alg: str, passphrase: bytes | None = None) -> bytes:
    private_key = serialization.load_pem_private_key(certificate, passphrase)
    if not isinstance(private_key, ec.EllipticCurvePrivateKey):
        raise TypeError("certificate must be an elliptic curve private key")

    return private_key.sign(data=data, signature_algorithm=ec.ECDSA(algorithm.get_hash_algorithm(hash_alg)))

verify 

verify(
    data: bytes,
    certificate: bytes,
    sig: bytes,
    hash_alg: str,
) -> bool
Source code in pypepper/common/security/crypto/elliptic/ecdsa.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
def verify(self, data: bytes, certificate: bytes, sig: bytes, hash_alg: str) -> bool:
    public_key = serialization.load_pem_public_key(certificate)
    if not isinstance(public_key, ec.EllipticCurvePublicKey):
        raise TypeError("certificate must be an elliptic curve public key")

    try:
        public_key.verify(
            signature=sig, data=data, signature_algorithm=ec.ECDSA(algorithm.get_hash_algorithm(hash_alg))
        )
    except exceptions.InvalidSignature:
        return False

    return True

get_private_key_pem staticmethod 

get_private_key_pem(
    private_key: EllipticCurvePrivateKey,
    passphrase: bytes | None = None,
) -> bytes
Source code in pypepper/common/security/crypto/elliptic/ecdsa.py 
37
38
39
40
41
42
43
44
45
46
47
48
@staticmethod
def get_private_key_pem(private_key: ec.EllipticCurvePrivateKey, passphrase: bytes | None = None) -> bytes:
    encryption_alg: KeySerializationEncryption
    if passphrase:
        encryption_alg = serialization.BestAvailableEncryption(passphrase)
    else:
        encryption_alg = serialization.NoEncryption()
    return private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=encryption_alg,
    )

get_public_key_pem staticmethod 

get_public_key_pem(
    private_key: EllipticCurvePrivateKey,
) -> bytes
Source code in pypepper/common/security/crypto/elliptic/ecdsa.py 
50
51
52
53
54
55
@staticmethod
def get_public_key_pem(private_key: ec.EllipticCurvePrivateKey) -> bytes:
    return private_key.public_key().public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo,
    )