Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import time
import _config as config
check_timeout = 180
bucket_fs_extra_timeout = 30
start_ts = time.time()
while True:
try:
C = pyexasol.connect(dsn=config.dsn, user=config.user, password=config.password)
print(f"Exasol was started in {time.time() - start_ts}s")
print(f"Sleeping extra {bucket_fs_extra_timeout}s to allow BucketFS to startup")
time.sleep(bucket_fs_extra_timeout)
break
except pyexasol.ExaConnectionError:
if (time.time() - start_ts) > check_timeout:
raise RuntimeError(f"Exasol did not start in {check_timeout}s, aborting test")
time.sleep(2)
"""
Example 9
Catching exceptions
"""
import pyexasol
import _config as config
import pprint
printer = pprint.PrettyPrinter(indent=4, width=140)
# Bad dsn
try:
C = pyexasol.connect(dsn='123' + config.dsn, user=config.user, password=config.password, schema=config.schema)
except pyexasol.ExaConnectionError as e:
print(e)
# Bad user \ password
try:
C = pyexasol.connect(dsn=config.dsn, user=config.user, password='123' + config.password, schema=config.schema)
except pyexasol.ExaAuthError as e:
print(e)
C = pyexasol.connect(dsn=config.dsn, user=config.user, password=config.password, schema=config.schema,
fetch_size_bytes=1024 * 10)
# Invalid SQL
try:
stmt = C.execute("""
SELECT1 *
FROM users