Skip to content

Instantly share code, notes, and snippets.

@Vizonex
Last active September 25, 2026 06:31
Show Gist options
  • Select an option

  • Save Vizonex/5a6979f7bedaafa4a7db438acda3d4bc to your computer and use it in GitHub Desktop.

Select an option

Save Vizonex/5a6979f7bedaafa4a7db438acda3d4bc to your computer and use it in GitHub Desktop.
Multidict Cython-API Generator is a generator for cython code. This may be a lot more opinionated then what was chosen for use with the multidict C-API.
# NOTE: Generated based off the verison that was written before the first release of the New Multidict C-API
# Sources may differ as updates keep getting produced.
from cpython.object cimport (
PyObject as PyObject,
PyTypeObject as PyTypeObject
)
from libc.stdint cimport uint64_t as uint64_t
cdef extern from "multidict_capi_struct.h":
ctypedef int (*MultiDict_ItemVisitor)(void *user_data, object key,
object value) noexcept
cdef extern from "multidict_capi.h":
"""
static MultiDict_CAPI* MultiDictAPI = NULL;
static inline int Pyx__Import_MultiDict(){
MultiDictAPI = MultiDict_GetCAPI();
if (MultiDictAPI == NULL){
return -1;
}
return 0;
}
static void __Pyx_Raise_MultiDict_NotImported_Error(){
PyErr_SetString(
PyExc_RuntimeError,
"multidict Cython C-API requires calling" \
"the \\"import_multidict\\" function."
);
}
#define __Pyx_Throw_NotImported_Error(ret) \
__Pyx_Raise_MultiDict_NotImported_Error(); \
return ret
static inline int Pyx__CIMultiDictProxy_Check(PyObject* op)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return CIMultiDictProxy_Check(MultiDictAPI, op);
}
static inline int Pyx__CIMultiDictProxy_CheckExact(PyObject* op)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return CIMultiDictProxy_CheckExact(MultiDictAPI, op);
}
static inline PyTypeObject* Pyx__CIMultiDictProxy_GetType()
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(NULL);
}
return CIMultiDictProxy_GetType(MultiDictAPI);
}
static inline PyObject* Pyx__CIMultiDictProxy_New(PyObject* arg)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(NULL);
}
return CIMultiDictProxy_New(MultiDictAPI, arg);
}
static inline int Pyx__CIMultiDict_Check(PyObject* op)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return CIMultiDict_Check(MultiDictAPI, op);
}
static inline int Pyx__CIMultiDict_CheckExact(PyObject* op)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return CIMultiDict_CheckExact(MultiDictAPI, op);
}
static inline PyTypeObject* Pyx__CIMultiDict_GetType()
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(NULL);
}
return CIMultiDict_GetType(MultiDictAPI);
}
static inline PyObject* Pyx__CIMultiDict_New(Py_ssize_t prealloc_size)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(NULL);
}
return CIMultiDict_New(MultiDictAPI, prealloc_size);
}
static inline int Pyx__IStr_Check(PyObject* op)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return IStr_Check(MultiDictAPI, op);
}
static inline int Pyx__IStr_CheckExact(PyObject* op)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return IStr_CheckExact(MultiDictAPI, op);
}
static inline PyObject* Pyx__IStr_FromUnicode(PyObject* str)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(NULL);
}
return IStr_FromUnicode(MultiDictAPI, str);
}
static inline PyTypeObject* Pyx__IStr_GetType()
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(NULL);
}
return IStr_GetType(MultiDictAPI);
}
static inline int Pyx__MultiDictProxy_Check(PyObject* op)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDictProxy_Check(MultiDictAPI, op);
}
static inline int Pyx__MultiDictProxy_CheckExact(PyObject* op)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDictProxy_CheckExact(MultiDictAPI, op);
}
static inline PyTypeObject* Pyx__MultiDictProxy_GetType()
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(NULL);
}
return MultiDictProxy_GetType(MultiDictAPI);
}
static inline PyObject* Pyx__MultiDictProxy_New(PyObject* arg)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(NULL);
}
return MultiDictProxy_New(MultiDictAPI, arg);
}
static inline int Pyx__MultiDict_Add(PyObject* self, PyObject* key, PyObject* value)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_Add(MultiDictAPI, self, key, value);
}
static inline int Pyx__MultiDict_Check(PyObject* op)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_Check(MultiDictAPI, op);
}
static inline int Pyx__MultiDict_CheckExact(PyObject* op)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_CheckExact(MultiDictAPI, op);
}
static inline int Pyx__MultiDict_Clear(PyObject* self)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_Clear(MultiDictAPI, self);
}
static inline int Pyx__MultiDict_Contains(PyObject* self, PyObject* key)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_Contains(MultiDictAPI, self, key);
}
static inline int Pyx__MultiDict_DelItem(PyObject* self, PyObject* key)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_DelItem(MultiDictAPI, self, key);
}
static inline Py_ssize_t Pyx__MultiDict_ForEach(PyObject* self, PyObject* key, MultiDict_ItemVisitor visitor, void* user_data)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_ForEach(MultiDictAPI, self, key, visitor, user_data);
}
static inline int Pyx__MultiDict_GetItem(PyObject* self, PyObject* key, PyObject** result)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_GetItem(MultiDictAPI, self, key, result);
}
static inline PyTypeObject* Pyx__MultiDict_GetType()
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(NULL);
}
return MultiDict_GetType(MultiDictAPI);
}
static inline uint64_t Pyx__MultiDict_GetVersion(PyObject* self)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(0);
}
return MultiDict_GetVersion(MultiDictAPI, self);
}
static inline PyObject* Pyx__MultiDict_New(Py_ssize_t prealloc_size)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(NULL);
}
return MultiDict_New(MultiDictAPI, prealloc_size);
}
static inline int Pyx__MultiDict_Pop(PyObject* self, PyObject* key, PyObject** result)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_Pop(MultiDictAPI, self, key, result);
}
static inline int Pyx__MultiDict_SetDefault(PyObject* self, PyObject* key, PyObject* default_value, PyObject** result)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_SetDefault(MultiDictAPI, self, key, default_value, result);
}
static inline int Pyx__MultiDict_SetItem(PyObject* self, PyObject* key, PyObject* value)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_SetItem(MultiDictAPI, self, key, value);
}
static inline Py_ssize_t Pyx__MultiDict_Size(PyObject* self)
{
if (MultiDictAPI == NULL){
__Pyx_Throw_NotImported_Error(-1);
}
return MultiDict_Size(MultiDictAPI, self);
}
"""
int import_multidict "Pyx__Import_MultiDict" () except -1
int CIMultiDictProxy_Check "Pyx__CIMultiDictProxy_Check" (object op) except -1
int CIMultiDictProxy_CheckExact "Pyx__CIMultiDictProxy_CheckExact" (object op) except -1
PyTypeObject* CIMultiDictProxy_GetType "Pyx__CIMultiDictProxy_GetType" () except NULL
object CIMultiDictProxy_New "Pyx__CIMultiDictProxy_New" (object arg)
int CIMultiDict_Check "Pyx__CIMultiDict_Check" (object op) except -1
int CIMultiDict_CheckExact "Pyx__CIMultiDict_CheckExact" (object op) except -1
PyTypeObject* CIMultiDict_GetType "Pyx__CIMultiDict_GetType" () except NULL
object CIMultiDict_New "Pyx__CIMultiDict_New" (Py_ssize_t prealloc_size)
int IStr_Check "Pyx__IStr_Check" (object op) except -1
int IStr_CheckExact "Pyx__IStr_CheckExact" (object op) except -1
object IStr_FromUnicode "Pyx__IStr_FromUnicode" (object str)
PyTypeObject* IStr_GetType "Pyx__IStr_GetType" () except NULL
int MultiDictProxy_Check "Pyx__MultiDictProxy_Check" (object op) except -1
int MultiDictProxy_CheckExact "Pyx__MultiDictProxy_CheckExact" (object op) except -1
PyTypeObject* MultiDictProxy_GetType "Pyx__MultiDictProxy_GetType" () except NULL
object MultiDictProxy_New "Pyx__MultiDictProxy_New" (object arg)
int MultiDict_Add "Pyx__MultiDict_Add" (object self, object key, object value) except -1
int MultiDict_Check "Pyx__MultiDict_Check" (object op) except -1
int MultiDict_CheckExact "Pyx__MultiDict_CheckExact" (object op) except -1
int MultiDict_Clear "Pyx__MultiDict_Clear" (object self) except -1
int MultiDict_Contains "Pyx__MultiDict_Contains" (object self, object key) except -1
int MultiDict_DelItem "Pyx__MultiDict_DelItem" (object self, object key) except -1
Py_ssize_t MultiDict_ForEach "Pyx__MultiDict_ForEach" (object self, object key, MultiDict_ItemVisitor visitor, void* user_data) except -1
int MultiDict_GetItem "Pyx__MultiDict_GetItem" (object self, object key, PyObject** result) except -1
PyTypeObject* MultiDict_GetType "Pyx__MultiDict_GetType" () except NULL
uint64_t MultiDict_GetVersion "Pyx__MultiDict_GetVersion" (object self) except *
object MultiDict_New "Pyx__MultiDict_New" (Py_ssize_t prealloc_size)
int MultiDict_Pop "Pyx__MultiDict_Pop" (object self, object key, PyObject** result) except -1
int MultiDict_SetDefault "Pyx__MultiDict_SetDefault" (object self, object key, object default_value, PyObject** result) except -1
int MultiDict_SetItem "Pyx__MultiDict_SetItem" (object self, object key, object value) except -1
Py_ssize_t MultiDict_Size "Pyx__MultiDict_Size" (object self) except -1
"""
Small Experiemntal System for generation of the Cython C-API branch.
This gets rid of the annoyance of needing to utilize a C-API Capsule
pointer and places that responsibility elsewhere. This also keeps
the system automatically updated to ensure the C-API is
up to date with the most recently made changes.
This system is inspired by numpy.
"""
import argparse
import os
import re
import sys
import urllib.request
from collections import OrderedDict
from dataclasses import dataclass
from pathlib import Path
CAPSULE_SOURCE = "multidict/multidict_capi.h"
DOWNLOAD_SOURCE_URL = "https://raw.githubusercontent.com/aio-libs/multidict/refs/heads/master/multidict/multidict_capi.h"
# C function signature regex, best method since
# utilizing libclang could just get too costly
# on diskspace some users may find needing libclang downright annoying.
FUNCTION_RE = re.compile(r"static inline ([^\s]+)\s+([^\(]+)\(([^\)]+)\)", flags=re.M)
# Prefix nickname for macros so that Compilation doesn't create
# confusion.
MACRO_NICKNAME = "Pyx_"
UNNESSESARY_FUNCTIONS = frozenset(["_MultiDict_CheckAPIVersion", "MultiDict_GetCAPI"])
MULTIDICT_API_CAPSULE = "MultiDictAPI"
CAPI_NON_EXISTANT_FUNC_NAME = "__Pyx_Raise_MultiDict_NotImported_Error"
CAPI_THROW_NOT_IMPORTED = "__Pyx_Throw_NotImported_Error"
CAPI_CYTHON_STRING_PRELUDE = f"""
static MultiDict_CAPI* {MULTIDICT_API_CAPSULE} = NULL;
static inline int Pyx__Import_MultiDict(){{
{MULTIDICT_API_CAPSULE} = MultiDict_GetCAPI();
if ({MULTIDICT_API_CAPSULE} == NULL){{
return -1;
}}
return 0;
}}
"""
# NOTE: {{}} is used to escape curly brackets
CAPI_NON_EXISTANT_FUNC = f"""
static void {CAPI_NON_EXISTANT_FUNC_NAME}(){{
PyErr_SetString(
PyExc_RuntimeError,
"multidict Cython C-API requires calling" \\
"the \\\\"import_multidict\\\\" function."
);
}}
#define {CAPI_THROW_NOT_IMPORTED}(ret) \\
{CAPI_NON_EXISTANT_FUNC_NAME}(); \\
return ret
"""
CYTHON_API_CODE_PRELUDE = """
from cpython.object cimport (
PyObject as PyObject,
PyTypeObject as PyTypeObject
)
from libc.stdint cimport uint64_t as uint64_t
cdef extern from "multidict_capi_struct.h":
ctypedef int (*MultiDict_ItemVisitor)(void *user_data, object key,
object value) noexcept
cdef extern from "multidict_capi.h":
\"\"\"
{}
\"\"\"
int import_multidict "Pyx__Import_MultiDict" () except -1
{}
"""
def cython_translation(ty: str):
match ty:
case "PyObject*":
return "object"
case _:
return ty
@dataclass
class CCall:
name: str
params: OrderedDict[str, str]
ret: str
@property
def cython_func(self) -> str:
return f"{MACRO_NICKNAME}_{self.name}"
@property
def filtered_params(self):
return OrderedDict([(k, v) for k, v in self.params.items() if k != "capi"])
def written_params(self):
return ", ".join(f"{ty} {name}" for name, ty in self.filtered_params.items())
def written_cython_params(self):
return ", ".join(
f"{cython_translation(ty)} {name}"
for name, ty in self.filtered_params.items()
)
def failure_object(self):
if self.ret == "uint64_t":
# use except *:... so cython can always check it.
return "0"
if "*" in self.ret:
return "NULL"
# Py_ssize_t or int
return "-1"
def cython_except(self):
if self.ret == "PyObject*":
# cython will handle it. Ignore...
return ""
if self.ret == "uint64_t":
return "except *"
if "*" in self.ret:
return "except NULL"
else:
return "except -1"
def c_function(self) -> str:
code = f"static inline {self.ret} {self.cython_func}"
code += "("
code += self.written_params()
code += ")\n"
code += "{\n"
code += f" if ({MULTIDICT_API_CAPSULE} == NULL){{\n"
code += f" {CAPI_THROW_NOT_IMPORTED}({self.failure_object()});\n"
code += " }\n"
params_line = ", ".join(self.filtered_params)
if not self.name.endswith(("_Check", "_CheckExact", "_GetType")):
code += f" return {self.name}({MULTIDICT_API_CAPSULE}, {params_line});\n}}\n"
return code
else:
if self.filtered_params:
code += f" return {self.name}({MULTIDICT_API_CAPSULE}, {params_line});\n}}\n"
else:
code += f" return {self.name}({MULTIDICT_API_CAPSULE});\n}}\n"
return code
def cython_signature(self) -> str:
code = f' {cython_translation(self.ret)} {self.name} "{self.cython_func}" ('
code += self.written_cython_params()
code += f") {self.cython_except()}\n"
return code
def parse_params(params: str):
if not params or params.strip() == "void":
# Empty function.
return OrderedDict()
# NOTE: Added Help for anybody confused.
# "int object, char* name" -> <("int object", "char* name")>
parameters = [p.strip() for p in params.split(",")]
# use rsplit incase of "unsigned char item" so the type is captured correctly.
name_and_type = [p.rsplit(" ", 1) for p in parameters]
# name and type are currently (ty, name) -> ("char*", "name") so they need reversing
# will then finalize the final product and return it, preserve ordering with OrderedDict
return OrderedDict([(name, ty) for ty, name in name_and_type])
def get_capi_calls(source: str) -> list[CCall]:
calls = []
for m in FUNCTION_RE.finditer(Path(source).read_text()):
ret, name, params_str = m.groups()
if name in UNNESSESARY_FUNCTIONS:
continue
calls.append(CCall(name, parse_params(params_str), ret))
return sorted(calls, key=lambda x: x.name)
def write_capi_cython_pxd(calls: list[CCall]):
C_FUNCTIONS = CAPI_CYTHON_STRING_PRELUDE + "\n" + CAPI_NON_EXISTANT_FUNC
C_FUNCTIONS += "\n".join([c.c_function() for c in calls])
CYTHON_PRELUDES = "\n".join([c.cython_signature() for c in calls])
return CYTHON_API_CODE_PRELUDE.format(C_FUNCTIONS, CYTHON_PRELUDES)
def get_source_from_venv():
"""Obtains source via virtual-enviornment (BEST Option.)"""
import multidict # noqa: PLC0415
return os.path.join(multidict.get_include(), "multidict_capi.h")
def main():
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"--download",
metavar="PATH",
help="downloads the most recently updated multidict c-api as source-code."
)
parser.add_argument(
"--source",
metavar="FILE.h",
default=CAPSULE_SOURCE,
help="Obtains multidict C-API source code, select --venv for internal use.",
)
parser.add_argument(
"--venv", "-v",
dest="venv",
action="store_true",
default=False,
help="obtains source-code over virtual enviornment."
)
parser.add_argument(
"--out",
"-o",
dest="out",
default=None,
metavar="FILE.c",
help="Output file to place genertated cython-capi signatures to.\n"
"Otherwise, Generated code is printed to the console.",
)
parser.add_argument(
"--verify",
action="store_true",
default=False,
help="verify that cython code is up to date with current C-API"
)
ns = parser.parse_args()
if download := ns.download:
target = os.path.join(download, DOWNLOAD_SOURCE_URL)
print(f"[...] Downloading file from {DOWNLOAD_SOURCE_URL}")
urllib.request.urlretrieve(DOWNLOAD_SOURCE_URL, target)
return
venv = None
if ns.venv:
print("[*] Virtual enviornment selected")
venv = get_source_from_venv()
output = ns.out
code = write_capi_cython_pxd(get_capi_calls(venv or ns.source or CAPSULE_SOURCE))
# generate code instead...
if not ns.verify:
if not output:
print("-- BEGIN CYTHON SOURCE --")
print(code)
print("-- END CYTHON SOURCE --")
else:
Path(output).write_text(code)
elif output:
has = {m.group(1) for m in re.finditer(r'"Pyx__([^"]+)"', Path(output).read_text())}
needed = {call.name for call in get_capi_calls(venv or ns.source or CAPSULE_SOURCE)}
# "Import_MultiDict" is a hard-coded function by default.
has.discard("Import_MultiDict")
if needed.difference(has):
print(f"[!] {output} is not up to date needed:")
for s in needed.difference(has):
print(f' - "{s}"')
sys.exit(-1)
else:
print(f"[+] {output} is up to date")
else:
print("[!] output argument is required when performing verification.")
sys.exit(-1)
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment