pyexasol/examples/16_ujson.py at master · 22quinn/pyexasol · GitHub

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
"""
Example 16
Edge case example with "ujson"
"""

import pyexasol
import _config as config

import decimal

import pprint
printer = pprint.PrettyPrinter(indent=4, width=140)

# Basic connect
C = pyexasol.connect(dsn=config.dsn, user=config.user, password=config.password, schema=config.schema, json_lib='ujson')

edge_cases = [
    # Biggest values
    {
        'dec36_0': decimal.Decimal('+' + ('9' * 36)),
        'dec36_36': decimal.Decimal('+0.' + ('9' * 36)),
        'dbl': 1.7e308,
        'bl': True,
        'dt': '9999-12-31',
        'ts': '9999-12-31 23:59:59.999',
        'var100': 'ひ' * 100,
        'var2000000': 'ひ' * 2000000,
    },
    # Smallest values
    {
        'dec36_0': decimal.Decimal('-' + ('9' * 36)),
        'dec36_36': decimal.Decimal('-0.' + ('9' * 36)),
        'dbl': -1.7e308,
        'bl': False,
        'dt': '0001-01-01',
        'ts': '0001-01-01 00:00:00',
        'var100': '',
        'var2000000': 'ひ',
    },
    # All nulls
    {
        'dec36_0': None,
        'dec36_36': None,
        'dbl': None,
        'bl': None,
        'dt': None,
        'ts': None,
        'var100': None,
        'var2000000': None,
    }
]

insert_q = 'INSERT INTO edge_case VALUES ({dec36_0!d}, {dec36_36!d}, {dbl!f}, {bl}, {dt}, {ts}, {var100}, {var2000000})'
select_q = 'SELECT dec36_0, dec36_36, dbl, bl, dt, ts, var100, LENGTH(var2000000) AS len_var FROM edge_case'

C.execute('TRUNCATE TABLE edge_case')

# Insert (test formatting)
C.execute(insert_q, dict(edge_cases[0]))
C.execute(insert_q, dict(edge_cases[1]))
C.execute(insert_q, dict(edge_cases[2]))

# Select and fetch
stmt = C.execute(select_q)
printer.pprint(stmt.fetchall())


# Same actions with "exasol_mapper"
C.fetch_mapper = pyexasol.exasol_mapper
C.execute('TRUNCATE TABLE edge_case')

# Insert (test formatting)
C.execute(insert_q, dict(edge_cases[0]))
C.execute(insert_q, dict(edge_cases[1]))
C.execute(insert_q, dict(edge_cases[2]))

# Select and fetch
stmt = C.execute(select_q)
printer.pprint(stmt.fetchall())

# Import and export
edge_tuples = C.execute("SELECT * FROM edge_case").fetchall()