我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用MySQLdb.paramstyle()。
def query(self, paramstyle=None): """ Returns the query part of the sql query. >>> q = SQLQuery(["SELECT * FROM test WHERE name=", SQLParam('joe')]) >>> q.query() 'SELECT * FROM test WHERE name=%s' >>> q.query(paramstyle='qmark') 'SELECT * FROM test WHERE name=?' """ s = [] for x in self.items: if isinstance(x, SQLParam): x = x.get_marker(paramstyle) s.append(safestr(x)) else: x = safestr(x) # automatically escape % characters in the query # For backward compatability, ignore escaping when the query looks already escaped if paramstyle in ['format', 'pyformat']: if '%' in x and '%%' not in x: x = x.replace('%', '%%') s.append(x) return "".join(s)
def __init__(self, **keywords): if 'pw' in keywords: keywords['password'] = keywords.pop('pw') db_module = import_driver(["psycopg2", "psycopg", "pgdb"], preferred=keywords.pop('driver', None)) if db_module.__name__ == "psycopg2": import psycopg2.extensions psycopg2.extensions.register_type(psycopg2.extensions.UNICODE) # if db is not provided postgres driver will take it from PGDATABASE environment variable if 'db' in keywords: keywords['database'] = keywords.pop('db') self.dbname = "postgres" self.paramstyle = db_module.paramstyle DB.__init__(self, db_module, keywords) self.supports_multiple_insert = True self._sequences = None
def __init__(self, **keywords): if 'pw' in keywords: keywords['password'] = keywords.pop('pw') db_module = import_driver(["psycopg2", "psycopg", "pgdb"], preferred=keywords.pop('driver', None)) if db_module.__name__ == "psycopg2": import psycopg2.extensions psycopg2.extensions.register_type(psycopg2.extensions.UNICODE) if db_module.__name__ == "pgdb" and 'port' in keywords: keywords["host"] += ":" + str(keywords.pop('port')) # if db is not provided postgres driver will take it from PGDATABASE environment variable if 'db' in keywords: keywords['database'] = keywords.pop('db') self.dbname = "postgres" self.paramstyle = db_module.paramstyle DB.__init__(self, db_module, keywords) self.supports_multiple_insert = True self._sequences = None
def aparam(): """ Returns the appropriate string to be used to interpolate a value with the current `web.ctx.db_module` or simply %s if there isn't one. >>> aparam() '%s' """ if hasattr(web.ctx, 'db_module'): style = web.ctx.db_module.paramstyle else: style = 'pyformat' if style == 'qmark': return '?' elif style == 'numeric': return ':1' elif style in ['format', 'pyformat']: return '%s' raise UnknownParamstyle, style
def _db_execute(self, cur, sql_query): """executes an sql query""" self.ctx.dbq_count += 1 try: a = time.time() paramstyle = getattr(self, 'paramstyle', 'pyformat') out = cur.execute(sql_query.query(paramstyle), [self._py2sql(x) for x in sql_query.values()]) b = time.time() except: if self.printing: print >> debug, 'ERR:', str(sql_query) if self.ctx.transactions: self.ctx.transactions[-1].rollback() else: self.ctx.rollback() raise if self.printing: print >> debug, '%s (%s): %s' % (round(b-a, 2), self.ctx.dbq_count, str(sql_query)) return out
def get_marker(self, paramstyle='pyformat'): if paramstyle == 'qmark': return '?' elif paramstyle == 'numeric': return ':1' elif paramstyle is None or paramstyle in ['format', 'pyformat']: return '%s' raise UnknownParamstyle, paramstyle
def _param_marker(self): """Returns parameter marker based on paramstyle attribute if this database.""" style = getattr(self, 'paramstyle', 'pyformat') if style == 'qmark': return '?' elif style == 'numeric': return ':1' elif style in ['format', 'pyformat']: return '%s' raise UnknownParamstyle, style
def _process_query(self, sql_query): """Takes the SQLQuery object and returns query string and parameters. """ paramstyle = getattr(self, 'paramstyle', 'pyformat') query = sql_query.query(paramstyle) params = sql_query.values() return query, params
def __init__(self, **keywords): db = import_driver(["sqlite3", "pysqlite2.dbapi2", "sqlite"], preferred=keywords.pop('driver', None)) if db.__name__ in ["sqlite3", "pysqlite2.dbapi2"]: db.paramstyle = 'qmark' # sqlite driver doesn't create datatime objects for timestamp columns unless `detect_types` option is passed. # It seems to be supported in sqlite3 and pysqlite2 drivers, not surte about sqlite. keywords.setdefault('detect_types', db.PARSE_DECLTYPES) self.paramstyle = db.paramstyle keywords['database'] = keywords.pop('db') self.dbname = "sqlite" DB.__init__(self, db, keywords)
def _process_query(self, sql_query): """Takes the SQLQuery object and returns query string and parameters. """ # MSSQLDB expects params to be a tuple. # Overwriting the default implementation to convert params to tuple. paramstyle = getattr(self, 'paramstyle', 'pyformat') query = sql_query.query(paramstyle) params = sql_query.values() return query, tuple(params)
def __init__(self, **keywords): import cx_Oracle as db if 'pw' in keywords: keywords['password'] = keywords.pop('pw') #@@ TODO: use db.makedsn if host, port is specified keywords['dsn'] = keywords.pop('db') self.dbname = 'oracle' db.paramstyle = 'numeric' self.paramstyle = db.paramstyle # oracle doesn't support pooling keywords.pop('pooling', None) DB.__init__(self, db, keywords)
def get_marker(self, paramstyle='pyformat'): if paramstyle == 'qmark': return '?' elif paramstyle == 'numeric': return ':1' elif paramstyle is None or paramstyle in ['format', 'pyformat']: return '%s' raise UnknownParamstyle(paramstyle)
def _param_marker(self): """Returns parameter marker based on paramstyle attribute if this database.""" style = getattr(self, 'paramstyle', 'pyformat') if style == 'qmark': return '?' elif style == 'numeric': return ':1' elif style in ['format', 'pyformat']: return '%s' raise UnknownParamstyle(style)
def __init__(self, **keywords): db = import_driver(["sqlite3", "pysqlite2.dbapi2", "sqlite"], preferred=keywords.pop('driver', None)) if db.__name__ in ["sqlite3", "pysqlite2.dbapi2"]: db.paramstyle = 'qmark' # sqlite driver doesn't create datatime objects for timestamp columns unless `detect_types` option is passed. # It seems to be supported in sqlite3 and pysqlite2 drivers, not surte about sqlite. keywords.setdefault('detect_types', db.PARSE_DECLTYPES) self.paramstyle = db.paramstyle keywords['database'] = keywords.pop('db') keywords['pooling'] = False # sqlite don't allows connections to be shared by threads self.dbname = "sqlite" DB.__init__(self, db, keywords)
def __init__(self, **keywords): try: import kinterbasdb as db except Exception: db = None pass if 'pw' in keywords: keywords['password'] = keywords.pop('pw') keywords['database'] = keywords.pop('db') self.paramstyle = db.paramstyle DB.__init__(self, db, keywords)
def __init__(self, **keywords): import MySQLdb as db if 'pw' in keywords: keywords['passwd'] = keywords['pw'] del keywords['pw'] if 'charset' not in keywords: keywords['charset'] = 'utf8' elif keywords['charset'] is None: del keywords['charset'] self.paramstyle = db.paramstyle = 'pyformat' # it's both, like psycopg self.dbname = "mysql" DB.__init__(self, db, keywords) self.supports_multiple_insert = True
def aparam(): """Use in a SQL string to make a spot for a db value.""" style = ctx.db_module.paramstyle if style == 'qmark': return '?' elif style == 'numeric': return ':1' elif style in ['format', 'pyformat']: return '%s' raise UnknownParamstyle, style
def query(self, paramstyle=None): """ Returns the query part of the sql query. >>> q = SQLQuery(["SELECT * FROM test WHERE name=", SQLParam('joe')]) >>> q.query() 'SELECT * FROM test WHERE name=%s' >>> q.query(paramstyle='qmark') 'SELECT * FROM test WHERE name=?' """ s = '' for x in self.items: if isinstance(x, SQLParam): x = x.get_marker(paramstyle) s += x return s
def __init__(self, **keywords): if 'pw' in keywords: keywords['password'] = keywords['pw'] del keywords['pw'] db_module = self.get_db_module() keywords['database'] = keywords.pop('db') self.dbname = "postgres" self.paramstyle = db_module.paramstyle DB.__init__(self, db_module, keywords) self.supports_multiple_insert = True
