# $SnapHashLicense:
#
# SnapLogic - Open source data services
#
# Copyright (C) 2008, SnapLogic, Inc. All rights reserved.
#
# See http://www.snaplogic.org for more information about
# the SnapLogic project.
#
# This program is free software, distributed under the terms of
# the GNU General Public License Version 2. See the LEGAL file
# at the top of the source tree.
#
# "SnapLogic" is a trademark of SnapLogic, Inc.
#
#
# $
# $Id: sqlite.py 5561 2008-12-02 00:25:19Z kurt $
from __future__ import with_statement
"""
SQLite store module.
This module contains the SQLite store object providing interfaces for SQLite specific operations.
"""
__docformat__ = "epytext en"
# Imports
import os, sys
from sqlite3 import dbapi2
import tempfile
import shutil
from snaplogic.server.repository.snap_sql_store import SnapSQLStore
from snaplogic.common.snap_exceptions import *
from snaplogic.common.db_utils import SQLQueryBuilder
from snaplogic.common.thread_utils import RWLock
# Public names
__all__ = [ "SQLite" ]
class SQLite(SnapSQLStore):
"""
SQLite store class. This class derives from SnapStore class and implemets SQLite specific operations
for the inherited interfaces.
"""
INSERT_OR_UPDATE_REGISTRY_ENTRY = SQLQueryBuilder("""REPLACE INTO registry VALUES (${name}, ${value})""")
"""
Timeout to wait for SQLite to obtain a lock on the database file (see ticket #1513).
"""
def __init__(self, rwlock=None):
"""
Initialization.
@param rwlock: A RWLock to use (when cloning).
@type rwlock: RWLock
"""
if rwlock is None:
rwlock = RWLock()
super(SQLite, self).__init__(dbapi2, rwlock)
self._db = None
@classmethod
def create(cls, path):
"""
Create and initialize a new SQLite database at path within the file system.
@param path: A file path to where the database should be created.
@type path: string
@raise SnapObjExistsError: A file or directory already exists at path.
@raise SnapRepInitError: An unexpected error occured while creating the database.
"""
db = None
try:
db = dbapi2.connect(path)
cursor = db.cursor()
cursor.execute('SELECT tbl_name FROM sqlite_master WHERE type="table"')
if len(cursor.fetchall()) > 0:
raise SnapObjExistsError("Cannot create SQLite database at '%s'; database already exists." % path)
cls._createTables(db, dbapi2)
except dbapi2.Error, e:
raise SnapException.chain(e,
SnapRepError("Error creating sqlite repository at '%s': %s." % (path, str(e))))
finally:
if db:
db.close()
@classmethod
def destroy(cls, path):
if not os.path.exists(path):
raise SnapObjNotFoundError("No SQLite database found at '%s'." % path)
db = None
try:
db = dbapi2.connect(path)
cls._destroy(db, dbapi2)
except Exception, e:
raise SnapException.chain(e, SnapRepError("Cannot connect to sqlite database at '%s'." % path))
finally:
if db:
db.close()
def connect(self, path):
"""
Connect to the SQLite database.
@param path: Path to the SQLite database file.
@type path: string
"""
if self._db is not None:
raise SnapRepError("Repository is already connected")
elif not os.path.exists(path):
raise SnapRepNotFoundError("SQLite database not found at '%s'." % path)
try:
self._db = dbapi2.connect(path)
self._db_path = path
except Exception, e:
raise SnapException.chain(e, SnapRepInitError("Cannot connect to sqlite database at '%s'." % path))
def close(self):
"""
Close the database.
"""
if self._db is not None:
self._db.close()
self._db = None
self._db_path = None
def clone(self):
instance = SQLite(self._lock)
instance.connect(self._db_path)
return instance
|