generated from dopt-python/py311
77 lines
1.7 KiB
Python
77 lines
1.7 KiB
Python
# %%
|
|
from pathlib import Path
|
|
|
|
import oracledb
|
|
import sqlalchemy as sql
|
|
from dopt_basics import configs, io
|
|
|
|
from umbreit import db
|
|
|
|
# %%
|
|
p_cfg = io.search_file_iterative(
|
|
starting_path=Path.cwd(),
|
|
glob_pattern="CRED*.toml",
|
|
stop_folder_name="umbreit-py",
|
|
)
|
|
assert p_cfg is not None
|
|
CFG = configs.load_toml(p_cfg)
|
|
HOST = CFG["server"]["host"]
|
|
PORT = CFG["server"]["port"]
|
|
SERVICE = CFG["server"]["service"]
|
|
USER_NAME = CFG["user"]["name"]
|
|
USER_PASS = CFG["user"]["pass"]
|
|
|
|
# %%
|
|
# !! init thick mode
|
|
# p_oracle_client = Path(r"C:\Databases\Oracle\instantclient_19_29")
|
|
# assert p_oracle_client.exists()
|
|
# assert p_oracle_client.is_dir()
|
|
# oracledb.init_oracle_client(lib_dir=str(p_oracle_client))
|
|
# %%
|
|
conn_string = (
|
|
f"oracle+oracledb://{USER_NAME}:{USER_PASS}@{HOST}:{PORT}?service_name={SERVICE}"
|
|
)
|
|
engine = sql.create_engine(conn_string)
|
|
# %%
|
|
# stmt = sql.select(db.EXT_BESPBES_INFO).limit(10)
|
|
stmt = sql.text("SELECT * FROM ext_bedpbed FETCH FIRST 30 ROWS ONLY")
|
|
compiled = stmt.compile(dialect=engine.dialect, compile_kwargs={"literal_binds": True})
|
|
print(str(compiled))
|
|
|
|
# %%
|
|
with engine.connect() as conn:
|
|
res = conn.execute(stmt)
|
|
|
|
res = tuple(res.all())
|
|
# %%
|
|
len(res[0])
|
|
# %%
|
|
res
|
|
|
|
# %%
|
|
engine.dispose()
|
|
|
|
# # %%
|
|
# stmt = sql.text("SELECT * FROM ALL_TAB_COLUMNS WHERE TABLE_NAME = 'ext_bedpbed' AND COLUMN_NAME = 'BEDARFNR'")
|
|
# with engine.connect() as conn:
|
|
# res = conn.execute(stmt)
|
|
# # %%
|
|
# res.all()
|
|
|
|
# %%
|
|
conn = oracledb.connect(
|
|
user=USER_NAME,
|
|
password=USER_PASS,
|
|
host=HOST,
|
|
port=PORT,
|
|
service_name=SERVICE,
|
|
)
|
|
# %%
|
|
cursor = conn.cursor()
|
|
cursor.execute("select * from ext_bedpbed limit 3")
|
|
# %%
|
|
print(cursor.fetchone())
|
|
# %%
|
|
conn.close()
|
|
# %%
|