forked from ebroder/python-hesiod
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy path_hesiod.pyx
91 lines (72 loc) · 2.41 KB
/
_hesiod.pyx
1
2
3
4
5
6
7
8
9
10
11
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
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
cdef extern from "hesiod.h":
int hesiod_init(void **context)
void hesiod_end(void *context)
char *hesiod_to_bind(void *context, char *name, char *type)
char **hesiod_resolve(void *context, char *name, char *type)
void hesiod_free_list(void *context, char **list)
# This function isn't defined in 3.0.2, which is what Debian/Ubuntu use
#void hesiod_free_string(void *context, char *str)
cdef extern from "errno.h":
int errno
cdef extern from "string.h":
char * strerror(int errnum)
cdef extern from "stdlib.h":
void free(void *)
cdef void * __context
cdef class __ContextManager:
def __init__(self):
if hesiod_init(&__context) == -1:
raise IOError(errno, strerror(errno))
def __del__(self):
hesiod_end(__context)
cdef object __cm
__cm = __ContextManager()
import threading
__lookup_lock = threading.Lock()
def bind(hes_name, hes_type):
"""
Convert the provided arguments into a DNS name.
The DNS name derived from the name and type provided is used to
actually execute the Hesiod lookup.
"""
cdef object py_result
cdef char * c_result
# N.B. libhesiod actually uses the locale, but
# locale.getpreferredencoding is not threadsafe so we have to
# assume utf-8.
if isinstance(hes_name, unicode):
hes_name = hes_name.encode('utf-8')
if isinstance(hes_type, unicode):
hes_type = hes_type.encode('utf-8')
c_result = hesiod_to_bind(__context, hes_name, hes_type)
if c_result is NULL:
raise IOError(errno, strerror(errno))
py_result = c_result
free(c_result)
return py_result
def resolve(hes_name, hes_type):
"""
Return a list of records matching the given name and type.
"""
cdef int i
cdef object py_result
py_result = list()
cdef int err = 0
cdef char ** c_result = NULL
if isinstance(hes_name, unicode):
hes_name = hes_name.encode('utf-8')
if isinstance(hes_type, unicode):
hes_type = hes_type.encode('utf-8')
with __lookup_lock:
c_result = hesiod_resolve(__context, hes_name, hes_type)
err = errno
if c_result is NULL:
raise IOError(err, strerror(err))
i = 0
while True:
if c_result[i] is NULL:
break
py_result.append(c_result[i].decode('utf-8', 'replace'))
i = i + 1
hesiod_free_list(__context, c_result)
return py_result