|
@@ -2,7 +2,7 @@ from typing import Dict, List, Union, Tuple, Optional, Any, Final, NamedTuple
|
|
|
import abc
|
|
import abc
|
|
|
from dataclasses import dataclass
|
|
from dataclasses import dataclass
|
|
|
import os
|
|
import os
|
|
|
-from collections import namedtuple
|
|
|
|
|
|
|
+from enum import Enum
|
|
|
|
|
|
|
|
import pandas as pd
|
|
import pandas as pd
|
|
|
|
|
|
|
@@ -19,15 +19,32 @@ class DBI(abc.ABC):
|
|
|
rows: int
|
|
rows: int
|
|
|
key: Optional[Any]
|
|
key: Optional[Any]
|
|
|
|
|
|
|
|
|
|
+ class ConfigSupplier(Enum):
|
|
|
|
|
+ FILE = 'FILE'
|
|
|
|
|
+ DIRECT = 'DIRECT'
|
|
|
|
|
+ ENV = 'ENV'
|
|
|
|
|
+
|
|
|
def __init__(self,
|
|
def __init__(self,
|
|
|
- host: str,
|
|
|
|
|
|
|
+ host: Optional[str] = None,
|
|
|
user: Optional[str] = None,
|
|
user: Optional[str] = None,
|
|
|
password: Optional[str] = None,
|
|
password: Optional[str] = None,
|
|
|
- scheme: Optional[str] = None):
|
|
|
|
|
- self.host: str = os.environ.get(host, 'localhost')
|
|
|
|
|
- self.user: Optional[str] = os.environ.get(user or '', None)
|
|
|
|
|
- self.password: Optional[str] = os.environ.get(password or '', None)
|
|
|
|
|
- self.scheme: Optional[str] = os.environ.get(scheme or '', None)
|
|
|
|
|
|
|
+ scheme: Optional[str] = None,
|
|
|
|
|
+ config_supply: ConfigSupplier = ConfigSupplier.ENV):
|
|
|
|
|
+ """Initialize a DBI class"""
|
|
|
|
|
+ self.OUTPUT(f'Using {config_supply.name} supplied DBI configuration', **self.OUTPUT_KWARGS)
|
|
|
|
|
+
|
|
|
|
|
+ if config_supply == DBI.ConfigSupplier.ENV:
|
|
|
|
|
+ self.host: Optional[str] = os.environ.get(host or '', 'localhost')
|
|
|
|
|
+ self.user: Optional[str] = os.environ.get(user or '', None)
|
|
|
|
|
+ self.password: Optional[str] = os.environ.get(password or '', None)
|
|
|
|
|
+ self.scheme: Optional[str] = os.environ.get(scheme or '', None)
|
|
|
|
|
+ elif config_supply == DBI.ConfigSupplier.FILE:
|
|
|
|
|
+ raise NotImplementedError()
|
|
|
|
|
+ else:
|
|
|
|
|
+ self.host = host
|
|
|
|
|
+ self.user = user
|
|
|
|
|
+ self.password = password
|
|
|
|
|
+ self.scheme = scheme
|
|
|
|
|
|
|
|
self.OUTPUT(f'Connecting to {self.host} -> {self.scheme if self.scheme else "*"}',
|
|
self.OUTPUT(f'Connecting to {self.host} -> {self.scheme if self.scheme else "*"}',
|
|
|
**self.OUTPUT_KWARGS)
|
|
**self.OUTPUT_KWARGS)
|