-
Notifications
You must be signed in to change notification settings - Fork 136
/
oracle.py
60 lines (47 loc) · 1.56 KB
/
oracle.py
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
from ..adapters.oracle import Oracle
import json
from .base import BasicParser, ListsParser
from datetime import datetime, date, time, timedelta
from base64 import b64decode
from .._compat import to_native
from . import parsers, for_type
class OracleParser(BasicParser):
@for_type("integer")
def _integer(self, value):
return int(value)
@for_type("text")
def _text(self, value):
return value
@for_type("clob")
def _clob(self, value):
return value
@for_type("blob")
def _blob(self, value):
decoded = b64decode(value.read())
try:
decoded = to_native(decoded)
except:
pass
return decoded
@for_type("json")
def _json(self, value):
return json.loads(value)
@for_type("date")
def _date(self, value):
if isinstance(value, datetime):
return value.date()
(y, m, d) = map(int, str(value)[:10].strip().split("-"))
return date(y, m, d)
@for_type("list:reference")
def _list_references(self, value, field_type):
return super(OracleParser, self)._list_references.f(self, value, field_type)
class OracleListsParser(ListsParser):
@for_type("list:integer")
def _list_integers(self, value):
return super(OracleListsParser, self)._list_integers.f(self, value)
@for_type("list:string")
def _list_strings(self, value):
return super(OracleListsParser, self)._list_strings.f(self, value)
@parsers.register_for(Oracle)
class OracleCommonparser(OracleParser, OracleListsParser):
pass