Source code for joule.utilities.connection_info

from typing import Dict
from joule.errors import ApiError

[docs] class ConnectionInfo: """ Encapsulates database connection information """ def __init__(self, username: str, password: str, port: int, database: str, host: str, nilmdb_url): """ Attributes: username (str): database username password (str): database password port (int): database port database (str): database name host (str): hostname nilmdb_url (str): NilmDB URL (empty if unused) """ self.username = username self.password = password self.port = port self.database = database = host self.nilmdb_url = nilmdb_url
[docs] def to_json(self) -> Dict: """ Convert connection parameters to a dictionary of attributes appropriate for transmission as JSON """ return { "username": self.username, "password": self.password, "port": self.port, "database": self.database, "host":, "nilmdb_url": self.nilmdb_url }
[docs] def to_dsn(self) -> str: """ Convert connection parameters to a DSN string used by SQLAlchemy and other database engines. """ return "postgresql://%s:%s@%s:%s/%s" % (self.username, self.password,, self.port, self.database)
def from_json(data: Dict) -> 'ConnectionInfo': # keep client backwards compatible with earlier jouled versions if "nilmdb_url" not in data: data["nilmdb_url"] = "" try: info = ConnectionInfo(username=data["username"], password=data["password"], port=data["port"], database=data["database"], host=data["host"], nilmdb_url=data["nilmdb_url"]) except KeyError as e: raise ApiError("invalid connection info, missing [%s]" % str(e)) return info