forked from apache/cassandra-python-driver
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdatatype_utils.py
More file actions
139 lines (110 loc) · 3.71 KB
/
datatype_utils.py
File metadata and controls
139 lines (110 loc) · 3.71 KB
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
92
93
94
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
# Copyright 2013-2014 DataStax, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from decimal import Decimal
import datetime
from uuid import UUID
import pytz
try:
from blist import sortedset
except ImportError:
sortedset = set # noqa
DATA_TYPE_PRIMITIVES = [
'ascii',
'bigint',
'blob',
'boolean',
# 'counter', counters are not allowed inside tuples
'decimal',
'double',
'float',
'inet',
'int',
'text',
'timestamp',
'timeuuid',
'uuid',
'varchar',
'varint',
]
DATA_TYPE_NON_PRIMITIVE_NAMES = [
'list',
'set',
'map',
'tuple'
]
def get_sample_data():
"""
Create a standard set of sample inputs for testing.
"""
sample_data = {}
for datatype in DATA_TYPE_PRIMITIVES:
if datatype == 'ascii':
sample_data[datatype] = 'ascii'
elif datatype == 'bigint':
sample_data[datatype] = 2 ** 63 - 1
elif datatype == 'blob':
sample_data[datatype] = bytearray(b'hello world')
elif datatype == 'boolean':
sample_data[datatype] = True
elif datatype == 'counter':
# Not supported in an insert statement
pass
elif datatype == 'decimal':
sample_data[datatype] = Decimal('12.3E+7')
elif datatype == 'double':
sample_data[datatype] = 1.23E+8
elif datatype == 'float':
sample_data[datatype] = 3.4028234663852886e+38
elif datatype == 'inet':
sample_data[datatype] = '123.123.123.123'
elif datatype == 'int':
sample_data[datatype] = 2147483647
elif datatype == 'text':
sample_data[datatype] = 'text'
elif datatype == 'timestamp':
sample_data[datatype] = datetime.datetime.fromtimestamp(872835240, tz=pytz.timezone('America/New_York')).astimezone(pytz.UTC).replace(tzinfo=None)
elif datatype == 'timeuuid':
sample_data[datatype] = UUID('FE2B4360-28C6-11E2-81C1-0800200C9A66')
elif datatype == 'uuid':
sample_data[datatype] = UUID('067e6162-3b6f-4ae2-a171-2470b63dff00')
elif datatype == 'varchar':
sample_data[datatype] = 'varchar'
elif datatype == 'varint':
sample_data[datatype] = int(str(2147483647) + '000')
else:
raise Exception('Missing handling of %s.' % datatype)
return sample_data
SAMPLE_DATA = get_sample_data()
def get_sample(datatype):
"""
Helper method to access created sample data
"""
return SAMPLE_DATA[datatype]
def get_nonprim_sample(non_prim_type, datatype):
"""
Helper method to access created sample data for non-primitives
"""
if non_prim_type == 'list':
return [get_sample(datatype), get_sample(datatype)]
elif non_prim_type == 'set':
return sortedset([get_sample(datatype)])
elif non_prim_type == 'map':
if datatype == 'blob':
return {get_sample('ascii'): get_sample(datatype)}
else:
return {get_sample(datatype): get_sample(datatype)}
elif non_prim_type == 'tuple':
return (get_sample(datatype),)
else:
raise Exception('Missing handling of non-primitive type {0}.'.format(non_prim_type))