Classes
class	connection

class	factory

Functions
def	_get_netrc_data (netrc_file, key)

def	connect (connection_data, mode="r", map_blobs=False, secrets=None, pooling=True)

def	engine_from_dictionary (dictionary, pooling=True)

def	new_connection_dictionary (connection_data, secrets=None, mode="r")

Detailed Description

connection class translates either a connection string for sqlite, oracle of frontier into a connection object.
Also sets up ORM with SQLAlchemy.

connection class can also take a pre-constructed engine - useful for web services.

Function Documentation

◆ _get_netrc_data()

def querying._get_netrc_data	(	netrc_file,
		key
	)

private

Returns a dictionary {login : ..., account : ..., password : ...}

Definition at line 337 of file querying.py.

References reco.zip().

Referenced by new_connection_dictionary().

 def _get_netrc_data(netrc_file, key):
     """
     Returns a dictionary {login : ..., account : ..., password : ...}
     """
     try:
         headers = ["login", "account", "password"]
         authenticator_tuple = netrc.netrc(netrc_file).authenticators(key)
         if authenticator_tuple == None:
             raise Exception("netrc file must contain key '%s'." % key)
     except:
         raise Exception("Couldn't get credentials from netrc file.")
     return dict(list(zip(headers, authenticator_tuple)))
 

◆ connect()

def querying.connect	(	connection_data,
		mode = `"r"`,
		map_blobs = `False`,
		secrets = `None`,
		pooling = `True`
	)

Utility method for user - set up a connection object.

Definition at line 453 of file querying.py.

Referenced by shell.connect(), command_line.copy_global_tag(), command_line.copy_tag(), command_line.diff_of_gts(), command_line.diff_of_tags(), command_line.list_object(), command_line.search(), uploads.uploader.send_payloads(), and payload_tests.payload_tests.test_write_blob_to_sqlite().

 def connect(connection_data, mode="r", map_blobs=False, secrets=None, pooling=True):
     """
     Utility method for user - set up a connection object.
     """
     con = connection(connection_data=connection_data, mode=mode, map_blobs=map_blobs, secrets=secrets, pooling=pooling)
     con = con.setup()
     return con

◆ engine_from_dictionary()

def querying.engine_from_dictionary	(	dictionary,
		pooling = `True`
	)

Definition at line 429 of file querying.py.

Referenced by querying.connection.setup().

 def engine_from_dictionary(dictionary, pooling=True):
     if dictionary["host"] != "sqlite":
         if dictionary["host"] != "frontier":
             # probably oracle
             # if not frontier, we have to authenticate
             user = dictionary["secrets"]["login"]
             pwd = dictionary["secrets"]["password"]
             # set max label length for oracle
             if pooling:
                 return create_engine(connection.build_oracle_url(user, pwd, dictionary["database_name"]), label_length=6)
             else:
                 return create_engine(connection.build_oracle_url(user, pwd, dictionary["database_name"]), label_length=6, poolclass=NullPool)
         else:
             # if frontier, no need to authenticate
             # set max label length for frontier
             if pooling:
                 return create_engine(connection.build_frontier_url(dictionary["database_name"], dictionary["schema"]), label_length=6)
             else:
                 return create_engine(connection.build_frontier_url(dictionary["database_name"], dictionary["schema"]), label_length=6, poolclass=NullPool)
     else:
         # if host is sqlite, making the url is easy - no authentication
         return create_engine("sqlite:///%s" % dictionary["database_name"])
 
 

◆ new_connection_dictionary()

def querying.new_connection_dictionary	(	connection_data,
		secrets = `None`,
		mode = `"r"`
	)

Function used to construct connection data dictionaries - internal to framework.

Definition at line 350 of file querying.py.

References _get_netrc_data(), input, submitPVValidationJobs.split(), and str.

 def new_connection_dictionary(connection_data, secrets=None, mode="r"):
     """
     Function used to construct connection data dictionaries - internal to framework.
     """
     frontier_str_length = len("frontier://")
     sqlite_str_length = len("sqlite://")
     #sqlite_file_str_length = len("sqlite_file://")
     oracle_str_length = len("oracle://")
 
     if type(connection_data) in [str, str] and connection_data[0:frontier_str_length] == "frontier://":
         """
         frontier://database_name/schema
         """
         db_name = connection_data[frontier_str_length:].split("/")[0]
         schema = connection_data[frontier_str_length:].split("/")[1]
         connection_data = {}
         connection_data["database_name"] = db_name
         connection_data["schema"] = schema
         connection_data["host"] = "frontier"
         connection_data["secrets"] = None
     elif type(connection_data) in [str, str] and connection_data[0:sqlite_str_length] == "sqlite://":
         """
         sqlite://database_file_name
         """
         # for now, just support "sqlite://" format for sqlite connection strings
         db_name = connection_data[sqlite_str_length:]
         schema = ""
         connection_data = {}
         connection_data["database_name"] = os.path.abspath(db_name)
         connection_data["schema"] = schema
         connection_data["host"] = "sqlite"
         connection_data["secrets"] = None
     elif type(connection_data) in [str, str] and connection_data[0:oracle_str_length] == "oracle://":
         """
         oracle://account:password@database_name
         or
         oracle://database_name/schema (requires a separate method of authentication - either dictionary or netrc)
         """
         new_connection_string = connection_data[oracle_str_length:]
 
         if ":" in new_connection_string:
             # the user has given a password - usually in the case of the db upload service
             database_name = new_connection_string[new_connection_string.index("@")+1:]
             schema_name = new_connection_string[0:new_connection_string.index(":")]
             # set username based on connection string
             username = new_connection_string[0:new_connection_string.index(":")]
             password = new_connection_string[new_connection_string.index(":")+1:new_connection_string.index("@")]
         else:
             mode_to_netrc_key_suffix = {"r" : "read", "w" : "write"}            database_name = new_connection_string[0:new_connection_string.index("/")]
             schema_name = new_connection_string[new_connection_string.index("/")+1:]
             if secrets == None:
                 username = str(input("Enter the username you want to connect to the schema '%s' with: " % (schema_name)))
                 password = str(input("Enter the password for the user '%s' in database '%s': " % (username, database_name)))
             else:
                 if type(secrets) == str:
                     netrc_key = "%s/%s/%s" % (database_name, schema_name, mode_to_netrc_key_suffix[mode])
                     netrc_data = _get_netrc_data(secrets, key=netrc_key)
                     # take the username from the netrc entry corresponding to the mode the database is opened in
                     # eg, if the user has given mode="read", the database_name/schema_name/read entry will be taken
                     username = netrc_data["login"]
                     password = netrc_data["password"]
                 elif type(secrets) == dict:
                     username = secrets["user"]
                     password = secrets["password"]
                 else:
                     raise Exception("Invalid type given for secrets.  Either an str or a dict must be given.")
 
         #print("Connected to database %s, schema %s, with username %s." % (database_name, schema_name, username))
 
         connection_data = {}
         connection_data["database_name"] = database_name
         connection_data["schema"] = schema_name
         connection_data["password"] = password
         connection_data["host"] = "oracle"
         connection_data["secrets"] = {"login" : username, "password" : password}
 
     return connection_data
 