/usr/share/pyshared/cyclone/sqlite.py is in python-cyclone 1.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | # coding: utf-8
#
# Copyright 2010 Alexandre Fiori
# based on the original Tornado by Facebook
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""An inline SQLite helper class.
All queries run inline, temporarily blocking the execution. Please make sure
you understand the limitations of using SQLite like this.
Example::
import cyclone.web
import cyclone.sqlite
class SQLiteMixin(object):
sqlite = cyclone.sqlite.InlineSQLite("mydb.sqlite")
class MyRequestHandler(cyclone.web.RequestHandler):
def get(self):
rs = self.sqlite.runQuery("SELECT 1")
...
There is no ``Deferred`` responses, and no need to ``yield`` anything.
"""
import sqlite3
class InlineSQLite:
"""An inline SQLite instance"""
def __init__(self, dbname=":memory:", autoCommit=True):
"""Create new SQLite instance."""
self.autoCommit = autoCommit
self.conn = sqlite3.connect(dbname)
self.curs = self.conn.cursor()
def runQuery(self, query, *args, **kwargs):
"""Use this function to execute queries that return a result set,
like ``SELECT``.
Example (with variable substitution)::
sqlite.runQuery("SELECT * FROM asd WHERE x=? and y=?", [x, y])
"""
self.curs.execute(query, *args, **kwargs)
return [row for row in self.curs]
def runOperation(self, command, *args, **kwargs):
"""Use this function to execute queries that do NOT return a result
set, like ``INSERT``, ``UPDATE`` and ``DELETE``.
Example::
sqlite.runOperation("CREATE TABLE asd (x int, y text)")
sqlite.runOperation("INSERT INTO asd VALUES (?, ?)", [x, y])
"""
self.curs.execute(command, *args, **kwargs)
if self.autoCommit is True:
self.conn.commit()
def runOperationMany(self, command, *args, **kwargs):
"""Same as `runOperation`, but for multiple rows.
Example::
sqlite.runOperationMany("INSERT INTO asd VALUES (?, ?)", [
[x1, y1], [x2, y2], [x3, y3]
])
"""
self.curs.executemany(command, *args, **kwargs)
if self.autoCommit is True:
self.conn.commit()
def commit(self):
"""Commits pending transactions"""
self.conn.commit()
def rollback(self):
"""Gives up pending transactions"""
self.conn.rollback()
def close(self):
"""Destroys the instance"""
self.conn.close()
|