atoti_jdbc.JdbcLoad

​

final class atoti_jdbc.JdbcLoad

>>> from atoti_jdbc import JdbcLoad
>>> url = f"h2:{database_path};USER=root;PASSWORD=pass"
>>> jdbc_load = JdbcLoad("SELECT * FROM MYTABLE", url=url)

>>> data_types = session.tables.infer_data_types(jdbc_load)
>>> data_types
{'ID': 'int', 'CITY': 'String', 'MY_VALUE': 'long'}

>>> table = session.create_table("Cities", data_types=data_types, keys={"ID"})

>>> table.load(jdbc_load)
>>> table.head().sort_index()
        CITY  MY_VALUE
ID
1      Paris       100
2     London        80
3   New York        90
4     Berlin        70
5    Jakarta        75

>>> table.drop()
>>> table.load(
...     JdbcLoad(
...         "SELECT * FROM MYTABLE WHERE City IN (?, ?)",
...         parameters=["Paris", "New York"],
...         url=url,
...     )
... )
>>> table.head().sort_index()
        CITY  MY_VALUE
ID
1      Paris       100
3   New York        90

>>> import sqlite3
>>> query = "SELECT * FROM MYTABLE"
>>> url = f"sqlite:{database_path}"
>>> with sqlite3.connect(database_path) as connection:
...     pd.read_sql(query, connection).set_index("ID")
        CITY    MY_VALUE
ID
1      Paris         100
2     London          80
3   New York  wrong type
4     Berlin          70
5    Jakarta          75

• With "log" or "skip":
  >>> table.drop()
  >>> table.load(JdbcLoad(query, error_handling="skip", url=url))
  >>> table.head().sort_index()
         CITY  MY_VALUE
  ID
  1     Paris       100
  2    London        80
  4    Berlin        70
  5   Jakarta        75
  
• With "warn":
  >>> table.drop()
  >>> import warnings
  >>> with warnings.catch_warnings(record=True) as warning_messages:
  ...     warnings.simplefilter("always")
  ...     table.load(JdbcLoad(query, error_handling="warn", url=url))
  >>> len(warning_messages)
  1
  >>> table.head().sort_index()
         CITY  MY_VALUE
  ID
  1     Paris       100
  2    London        80
  4    Berlin        70
  5   Jakarta        75
  
• With "error":
  >>> table.drop()
  >>> table.load(
  ...     JdbcLoad(query, error_handling="fail", url=url)
  ... )  
  Traceback (most recent call last):
      ...
  py4j.protocol.Py4JError: ...
  
  The content of the table after an invalid row occurred is undefined: some rows may or may not have been loaded. Use data_transaction() to keep the operation atomic:
  >>> table.drop()
  >>> table += (0, "Rome", 42)
  >>> with session.tables.data_transaction():  
  ...     table.load(JdbcLoad(query, error_handling="fail", url=url))
  Traceback (most recent call last):
      ...
  py4j.protocol.Py4JError: ...
  >>> table.head()
      CITY  MY_VALUE
  ID
  0   Rome        42
  

​

driver : Annotated[str | None, AfterValidator(_validate_driver)] = None

​

error_handling : ErrorHandling = ‘fail’

• "fail": raise an error on the first invalid row and interrupt the load.
• "warn": skip invalid rows and raise a warning if any was found.
• "log": skip invalid rows and log each of them.
• "skip": skip invalid rows without logging. Health events are still dispatched for observability.

​

parameters : FrozenSequence[Constant] = ()

​

query : str

​

url : Annotated[str, AfterValidator(normalize_jdbc_url)]

• "h2:/home/user/database/file/path;USER=username;PASSWORD=passwd"
• "postgresql://postgresql.db.server:5430/example?user=username&password=passwd"