PerfDB Class¶
PerfDB(user=None, password=None, database=None, host=None, schema=None, application_name=None, **kwargs)
¶
Top level class used to connect to PerformanceDB. It will create the connection to the database and then allow for direct access to it's content.
By default no arguments needs to be provided as the package will use the default ones, but its highly recommended to provide an application name.
Parameters:
-
(user¶str | None, default:None) –Name of PerformanceDB user. If None, default will be used. By default None
-
(password¶str | None, default:None) –PerformanceDB user password. If None, default will be used. By default None
-
(database¶str | None, default:None) –PerformanceDB database name. If None, default will be used. By default None
-
(host¶str | None, default:None) –PerformanceDB host (IP address). If None, default will be used. By default None
-
(schema¶str | None, default:None) –PerformanceDB schema name. If None, default will be used. By default None
-
(application_name¶str | None, default:None) –Application name to be used in the connection. This name will show up in pgAdmin while the connection is active. If None, default will be used. By default None
Other Parameters:
-
port(int) –Port number to connect to. By default 5432
-
conn_timeout(int) –Connection timeout in seconds. By default 3
-
query_timeout(int) –Query timeout in seconds. By default 300
-
autocommit(bool) –Autocommit mode. By default True
-
max_retries(int) –Maximum number of retries in case of connection errors. By default 3
-
retry_wait_time(int) –Time to wait between retries in seconds. By default 0.5
-
time_zone(str) –Time zone to be used in the connection. By default "America/Sao_Paulo".
Source code in echo_postgres/perfdb.py
@validate_call
def __init__(
self,
user: str | None = None,
password: str | None = None,
database: str | None = None,
host: str | None = None,
schema: str | None = None,
application_name: str | None = None,
**kwargs: Unpack[PerfDBKwargs],
) -> None:
"""
Top level class used to connect to PerformanceDB. It will create the connection to the database and then allow for direct access to it's content.
By default no arguments needs to be provided as the package will use the default ones, but its highly recommended to provide an application name.
Parameters
----------
user : str | None, optional
Name of PerformanceDB user. If None, default will be used. By default None
password : str | None, optional
PerformanceDB user password. If None, default will be used. By default None
database : str | None, optional
PerformanceDB database name. If None, default will be used. By default None
host : str | None, optional
PerformanceDB host (IP address). If None, default will be used. By default None
schema : str | None, optional
PerformanceDB schema name. If None, default will be used. By default None
application_name : str | None, optional
Application name to be used in the connection. This name will show up in pgAdmin while the connection is active.
If None, default will be used. By default None
Other Parameters
----------------
port : int, optional
Port number to connect to. By default 5432
conn_timeout : int, optional
Connection timeout in seconds. By default 3
query_timeout : int, optional
Query timeout in seconds. By default 300
autocommit : bool, optional
Autocommit mode. By default True
max_retries : int, optional
Maximum number of retries in case of connection errors. By default 3
retry_wait_time : int, optional
Time to wait between retries in seconds. By default 0.5
time_zone : str, optional
Time zone to be used in the connection. By default "America/Sao_Paulo".
"""
# creating connection properties to be used by the handler
conn_props = SqlConnProperties(
host=host or DEFAULT_HOST,
user=user or DEFAULT_USER,
password=password or DEFAULT_PASSWORD,
database=database or DEFAULT_DATABASE,
schema=schema or DEFAULT_SCHEMA,
port=kwargs.get("port", DEFAULT_PORT),
conn_timeout=kwargs.get("conn_timeout", 3),
query_timeout=kwargs.get("query_timeout", 300),
autocommit=kwargs.get("autocommit", True),
application_name=application_name,
time_zone=kwargs.get("time_zone", DEFAULT_TIME_ZONE),
)
# creating connection handler
# skip connect is used to avoid trying to connect when creating the handler
# connection will be done on each method call
self.conn = PgSqlHandler(
connection_properties=conn_props,
max_retries=kwargs.get("max_retries", 3),
retry_wait_time=kwargs.get("retry_wait_time", 0.5),
skip_connect=True,
)
from .alarms import Alarms
from .attributes import Attributes
from .bazefieldusers import BazefieldUsers
from .calcmodels import CalcModels
from .ccee import Ccee
from .comments import Comments
from .companies import Companies
from .components import Components
from .datasources import DataSources
from .datatypes import DataTypes
from .documents import Documents
from .events import Events
from .features import Features
from .forecasts import Forecasts
from .jobs import Jobs
from .kpis import Kpis
from .labels import Labels
from .objects import Objects
from .ons import Ons
from .rawdata import RawData
from .resourceassessments import ResourceAssessments
from .rootcauses import RootCauses
from .settings import Settings
from .users import Users
from .vibration import Vibration
# * subclasses
self.alarms = Alarms(self)
self.attributes = Attributes(self)
self.bazefieldusers = BazefieldUsers(self)
self.calcmodels = CalcModels(self)
self.ccee = Ccee(self)
self.comments = Comments(self)
self.companies = Companies(self)
self.components = Components(self)
self.datasources = DataSources(self)
self.datatypes = DataTypes(self)
self.documents = Documents(self)
self.events = Events(self)
self.features = Features(self)
self.forecasts = Forecasts(self)
self.jobs = Jobs(self)
self.kpis = Kpis(self)
self.labels = Labels(self)
self.objects = Objects(self)
self.ons = Ons(self)
self.rawdata = RawData(self)
self.resourceassessments = ResourceAssessments(self)
self.rootcauses = RootCauses(self)
self.settings = Settings(self)
self.users = Users(self)
self.vibration = Vibration(self)