Lesson 05
PluginClass = RegistryMeta.get_plugin('MyPlugin')OOP, async, and HTTP
plugin = PluginClass()Design with classes, embrace async for I/O, and talk to APIs safely.
print(plugin.run('hello world')) # dlrow ollehAdvanced Coding Tutorials: Expert!
Section 1: Metaclasses & Dynamic Class Creation +
Explanation: Here we define a metaclass that automatically registers every subclass in a global registry. This allows dynamic lookup and plugin architectures without explicit imports.
Skip to content
class RegistryMeta(type):
SSigLabs Code _registry = {}
@classmethod
def get_plugin(mcs, name):
return mcs._registry[name]
return data[::-1]
# Usage
Section 2: Bytecode Manipulation & Runtime Patching +
Explanation: We load a function’s bytecode, modify opcodes on the fly (e.g., inject logging at entry), rebuild the code object, and replace the original. This demands deep understanding of Python’s bytecode and the types.CodeType signature.
class Instrument:
def __init__(self, symbol: str, price: float):
self.symbol = symbolimport dis, types
self.price = price
def trace_entry(func):
def update(self, price: float) -> None: co = func.__code__
self.price = price instructions = list(dis.Bytecode(co))
new_insts = []
sig = Instrument("SIG", 12.34)
sig.update(13.05) # Inject a PRINT_OP at start
print(sig.symbol, sig.price) new_insts.append(dis.Instruction('LOAD_GLOBAL', co.co_names.index('print'), None, None, None, None, None))
new_insts.append(dis.Instruction('LOAD_CONST', co.co_consts.index(f'Entering {func.__name__}'), None, None, None, None, None))
new_insts.append(dis.Instruction('CALL_FUNCTION', 1, None, None, None, None, None))
dataclasses
new_insts.append(dis.Instruction('POP_TOP', None, None, None, None, None, None))
from dataclasses import dataclass new_insts.extend(instructions)
@dataclass bytecode = b''.join([inst.to_bytes() for inst in new_insts])
class Quote: new_code = types.CodeType(
symbol: str co.co_argcount,
price: float co.co_posonlyargcount,
co.co_kwonlyargcount,
q = Quote("SIG", 12.34) co.co_nlocals,
print(q) co.co_stacksize + 2,
co.co_flags,
bytecode,
Async/await
co.co_consts,
import asyncio co.co_names,
co.co_varnames,
async def fetch_price(symbol: str) -> float: co.co_filename,
await asyncio.sleep(0.1) # simulate I/O co.co_name,
return 12.34 co.co_firstlineno,
co.co_lnotab,
async def main(): co.co_freevars,
prices = await asyncio.gather(*(fetch_price(s) for s in ["SIG", "EURUSD"])) co.co_cellvars
print(prices) )
func.__code__ = new_code
asyncio.run(main()) return func
@trace_entry
HTTP requests
def complex_calc(x, y):
import requests return x * y + (x - y)
resp = requests.get("https://httpbin.org/get", timeout=10)print(complex_calc(10, 5))
resp.raise_for_status()
data = resp.json()
Section 3: Writing a Custom C Extension Module +
Explanation: This shows the boilerplate for a C extension exposing a high‑performance `matrix_multiply` to Python. It involves managing PyObject references and ensuring thread safety via the GIL.
print(data["url"])
// matrixmodule.c
#include
Accelerate with SigLabs Financial Software
Forecasting, market screening, risk insights, and reporting—built for speed and clarity.
static PyObject* matmul(PyObject* self, PyObject* args) {
Py_buffer A, B;
if (!PyArg_ParseTuple(args, "y*y*", &A, &B)) return NULL;
// Assume square N×N, pointer arithmetic, fast loops...
Next steps
double* a = (double*)A.buf;
double* b = (double*)B.buf;
← Prev: Files, errors, and testing Py_buffer C;
All lessons PyBuffer_FillInfo(&C, NULL, malloc(A.len), A.len, 1, PyBUF_CONTIG);
double* c = (double*)C.buf;
int N = (int)sqrt(A.len / sizeof(double));
for (int i=0; i
Section 4: Asyncio Internals & Custom Event Loop +
Explanation: Subclass the default selector event loop to insert a custom scheduling strategy (priority queue) and integrate low-level watcher callbacks without blocking the loop.
class PriorityEventLoop(asyncio.SelectorEventLoop):
def __init__(self):
super().__init__()