Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# SPDX-FileCopyrightText: 2014-2021 SAP SE
#
# SPDX-License-Identifier: Apache-2.0
from hdbcli import dbapi
conn = dbapi.connect(
address="localhost",
port=50000,
user="SYSTEM",
password="Toor1234"
)
cursor = conn.cursor()
cursor.execute("DROP TABLE T1")
cursor.execute("CREATE TABLE T1 (ID INTEGER PRIMARY KEY, C2 VARCHAR(255))")
cursor.close()
sql = 'INSERT INTO T1 (ID, C2) VALUES (?, ?)'
cursor = conn.cursor()
cursor.execute(sql, (1, 'hello'))
Args:
host (str): Host where the database is running
port (int): Database port (3{inst_number}15 by default)
user (str): Existing username in the database
password (str): User password
properties : Additional properties can be used with named parameters. More info at:
https://help.sap.com/viewer/0eec0d68141541d1b07893a39944924e/2.0.02/en-US/ee592e89dcce4480a99571a4ae7a702f.html
Example:
To avoid automatic reconnection set RECONNECT='FALSE' as parameter
"""
self._logger.info('connecting to SAP HANA database at %s:%s', host, port)
self.__properties = kwargs
try:
self._connection = dbapi.connect(
address=host,
port=port,
#user=kwargs.get('user'),
#password=kwargs.get('password'),
**self.__properties
)
except dbapi.Error as err:
raise base_connector.ConnectionError('connection failed: {}'.format(err))
self._logger.info('connected successfully')
self.cursor = self.db.cursor()
elif self.dbtype == "oracle":
s = "%s/%s@(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=%s)(PORT=%s))(CONNECT_DATA=(SERVICE_NAME=%s)))"
s = s % (user, pwd, hostname, port, dbname)
self.db = cx_Oracle.connect(s)
self.cursor = self.db.cursor()
elif self.dbtype == "netezza":
# conn="DRIVER={MySQL ODBC 3.51 Driver}; SERVER=localhost; PORT=3306; DATABASE=mysql; UID=joe;
# PASSWORD=bloggs; OPTION=3;SOCKET=/var/run/mysqld/mysqld.sock;"
self.cursor = Connect(hostname, user, pwd)
elif self.dbtype in ["hana"]:
from hdbcli import dbapi
self.db = dbapi.connect(
address=hostname, port=30015+int(port), user=user, password=pwd, autocommit=True)
self.cursor = self.db.cursor()
elif self.dbtype in ["progress"]:
dsn = self.odbc(hostname, port, dbname)
self.db = pyodbc.connect(
dsn=dsn, user=user, password=pwd, autocommit=True)
self.cursor = self.db.cursor()
elif self.dbtype in ["zen"]:
# Example: driver={Pervasive ODBC Interface};server=localhost;DBQ=demodata'
# Example: driver={Pervasive ODBC Interface};server=hostname:port;serverdsn=dbname'
dsn = dbname
connString = "DRIVER={Pervasive ODBC Interface};SERVER=%s;ServerDSN=%s;UID=%s;PWD=%s;" % (
hostname, dsn, user, pwd)
if connect:
def dbapi(cls):
import hdbcli.dbapi
hdbcli.dbapi.paramstyle = cls.default_paramstyle
return hdbcli.dbapi