36 lines
929 B
Python
36 lines
929 B
Python
from sqlalchemy import create_engine, Text
|
|
import pandas as pd
|
|
from dotenv import load_dotenv
|
|
from pathlib import Path
|
|
from urllib.parse import quote
|
|
import os
|
|
import logging
|
|
|
|
|
|
env_path = Path("config/settings.env")
|
|
load_dotenv(env_path)
|
|
|
|
oracle_conn_str = os.getenv("oracle_conn_str")
|
|
co_dw_conn_str = os.getenv("co_dw_conn_str")
|
|
co_daten_conn_str = os.getenv("co_daten_conn_str")
|
|
|
|
def get_conn(db):
|
|
match db:
|
|
case "oracle":
|
|
engine = create_engine(oracle_conn_str)
|
|
case "co_dw":
|
|
engine = create_engine(co_dw_conn_str)
|
|
case "co_daten":
|
|
engine = create_engine(co_dw_conn_str)
|
|
case _:
|
|
logging.info(f"Datenbank {db} konnte nicht gefunden werden")
|
|
return engine
|
|
|
|
# def get_data(db):
|
|
# engine = get_conn(db)
|
|
# with engine.connect() as conn:
|
|
# print(engine)
|
|
# return
|
|
|
|
# if __name__ == "__main__":
|
|
# get_data("co_daten") |