1 """provide an abstract class for external sources using a sqlite database helper

     2 

     3 :organization: Logilab

     4 :copyright: 2007-2008 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

     5 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr

     6 """

     7 __docformat__ = "restructuredtext en"

     8 

     9 

    10 import time

    11 import threading

    12 from os.path import join, exists

    13 

    14 from cubicweb import server

    15 from cubicweb.server.sqlutils import sqlexec, SQLAdapterMixIn

    16 from cubicweb.server.sources import AbstractSource, native

    17 from cubicweb.server.sources.rql2sql import SQLGenerator

    18 

    19 def timeout_acquire(lock, timeout):

    20     while not lock.acquire(False):

    21         time.sleep(0.2)

    22         timeout -= 0.2

    23         if timeout <= 0:

    24             raise RuntimeError("svn source is busy, can't acquire connection lock")

    25         

    26 class ConnectionWrapper(object):

    27     def __init__(self, source=None):

    28         self.source = source

    29         self._cnx = None

    30 

    31     @property

    32     def cnx(self):

    33         if self._cnx is None:

    34             timeout_acquire(self.source._cnxlock, 5)

    35             self._cnx = self.source._sqlcnx

    36         return self._cnx

    37     

    38     def commit(self):

    39         if self._cnx is not None:

    40             self._cnx.commit()

    41         

    42     def rollback(self):

    43         if self._cnx is not None:

    44             self._cnx.rollback()

    45         

    46     def cursor(self):

    47         return self.cnx.cursor()

    48 

    49     

    50 class SQLiteAbstractSource(AbstractSource):

    51     """an abstract class for external sources using a sqlite database helper

    52     """

    53     sqlgen_class = SQLGenerator

    54     @classmethod

    55     def set_nonsystem_types(cls):

    56         # those entities are only in this source, we don't want them in the

    57         # system source

    58         for etype in cls.support_entities:

    59             native.NONSYSTEM_ETYPES.add(etype)

    60         for rtype in cls.support_relations:

    61             native.NONSYSTEM_RELATIONS.add(rtype)

    62         

    63     options = (

    64         ('helper-db-path',

    65          {'type' : 'string',

    66           'default': None,

    67           'help': 'path to the sqlite database file used to do queries on the \

    68 repository.',

    69           'inputlevel': 2,

    70           }),

    71     )

    72             

    73     def __init__(self, repo, appschema, source_config, *args, **kwargs):

    74         # the helper db is used to easy querying and will store everything but

    75         # actual file content 

    76         dbpath = source_config.get('helper-db-path')

    77         if dbpath is None:

    78             dbpath = join(repo.config.appdatahome,

    79                           '%(uri)s.sqlite' % source_config)

    80         self.dbpath = dbpath

    81         self.sqladapter = SQLAdapterMixIn({'db-driver': 'sqlite',

    82                                            'db-name': dbpath})

    83         # those attributes have to be initialized before ancestor's __init__

    84         # which will call set_schema

    85         self._need_sql_create = not exists(dbpath)

    86         self._need_full_import = self._need_sql_create

    87         AbstractSource.__init__(self, repo, appschema, source_config,

    88                                 *args, **kwargs)

    89         # sql database can only be accessed by one connection at a time, and a

    90         # connection can only be used by the thread which created it so:

    91         # * create the connection when needed

    92         # * use a lock to be sure only one connection is used

    93         self._cnxlock = threading.Lock()

    94         

    95     @property

    96     def _sqlcnx(self):

    97         # XXX: sqlite connections can only be used in the same thread, so

    98         #      create a new one each time necessary. If it appears to be time

    99         #      consuming, find another way

   100         return self.sqladapter.get_connection()

   101 

   102     def _is_schema_complete(self):

   103         for etype in self.support_entities:

   104             if not etype in self.schema:

   105                 self.warning('not ready to generate %s database, %s support missing from schema',

   106                              self.uri, etype)

   107                 return False

   108         for rtype in self.support_relations:

   109             if not rtype in self.schema:

   110                 self.warning('not ready to generate %s database, %s support missing from schema',

   111                              self.uri, rtype)

   112                 return False

   113         return True

   114 

   115     def _create_database(self):

   116         from yams.schema2sql import eschema2sql, rschema2sql

   117         from cubicweb.toolsutils import restrict_perms_to_user

   118         self.warning('initializing sqlite database for %s source' % self.uri)

   119         cnx = self._sqlcnx

   120         cu = cnx.cursor()

   121         schema = self.schema

   122         for etype in self.support_entities:

   123             eschema = schema.eschema(etype)

   124             createsqls = eschema2sql(self.sqladapter.dbhelper, eschema,

   125                                      skip_relations=('data',))

   126             sqlexec(createsqls, cu, withpb=False)

   127         for rtype in self.support_relations:

   128             rschema = schema.rschema(rtype)

   129             if not rschema.inlined:

   130                 sqlexec(rschema2sql(rschema), cu, withpb=False)

   131         cnx.commit()

   132         cnx.close()

   133         self._need_sql_create = False

   134         if self.repo.config['uid']:

   135             from logilab.common.shellutils import chown

   136             # database file must be owned by the uid of the server process

   137             self.warning('set %s as owner of the database file',

   138                          self.repo.config['uid'])

   139             chown(self.dbpath, self.repo.config['uid'])

   140         restrict_perms_to_user(self.dbpath, self.info)

   141         

   142     def set_schema(self, schema):

   143         super(SQLiteAbstractSource, self).set_schema(schema)

   144         if self._need_sql_create and self._is_schema_complete():

   145             self._create_database()

   146         self.rqlsqlgen = self.sqlgen_class(schema, self.sqladapter.dbhelper)

   147                 

   148     def get_connection(self):

   149         return ConnectionWrapper(self)

   150 

   151     def check_connection(self, cnx):

   152         """check connection validity, return None if the connection is still valid

   153         else a new connection (called when the pool using the given connection is

   154         being attached to a session)

   155 

   156         always return the connection to reset eventually cached cursor

   157         """

   158         return cnx

   159 

   160     def pool_reset(self, cnx):

   161         """the pool using the given connection is being reseted from its current

   162         attached session: release the connection lock if the connection wrapper

   163         has a connection set

   164         """

   165         if cnx._cnx is not None:

   166             try:

   167                 cnx._cnx.close()

   168                 cnx._cnx = None

   169             finally:

   170                 self._cnxlock.release()

   171         

   172     def syntax_tree_search(self, session, union,

   173                            args=None, cachekey=None, varmap=None, debug=0):

   174         """return result from this source for a rql query (actually from a rql 

   175         syntax tree and a solution dictionary mapping each used variable to a 

   176         possible type). If cachekey is given, the query necessary to fetch the

   177         results (but not the results themselves) may be cached using this key.

   178         """

   179         if self._need_sql_create:

   180             return []

   181         sql, query_args = self.rqlsqlgen.generate(union, args)

   182         if server.DEBUG:

   183             print self.uri, 'SOURCE RQL', union.as_string()

   184             print 'GENERATED SQL', sql

   185         args = self.sqladapter.merge_args(args, query_args)

   186         cursor = session.pool[self.uri]

   187         cursor.execute(sql, args)

   188         return self.sqladapter.process_result(cursor) 

   189 

   190     def local_add_entity(self, session, entity):

   191         """insert the entity in the local database.

   192 

   193         This is not provided as add_entity implementation since usually source

   194         don't want to simply do this, so let raise NotImplementedError and the

   195         source implementor may use this method if necessary

   196         """

   197         cu = session.pool[self.uri]

   198         attrs = self.sqladapter.preprocess_entity(entity)

   199         sql = self.sqladapter.sqlgen.insert(str(entity.e_schema), attrs)

   200         cu.execute(sql, attrs)

   201         

   202     def add_entity(self, session, entity):

   203         """add a new entity to the source"""

   204         raise NotImplementedError()

   205 

   206     def local_update_entity(self, session, entity):

   207         """update an entity in the source

   208 

   209         This is not provided as update_entity implementation since usually

   210         source don't want to simply do this, so let raise NotImplementedError

   211         and the source implementor may use this method if necessary

   212         """

   213         cu = session.pool[self.uri]

   214         attrs = self.sqladapter.preprocess_entity(entity)

   215         sql = self.sqladapter.sqlgen.update(str(entity.e_schema), attrs, ['eid'])

   216         cu.execute(sql, attrs)

   217         

   218     def update_entity(self, session, entity):

   219         """update an entity in the source"""

   220         raise NotImplementedError()

   221         

   222     def delete_entity(self, session, etype, eid):

   223         """delete an entity from the source

   224 

   225         this is not deleting a file in the svn but deleting entities from the

   226         source. Main usage is to delete repository content when a Repository

   227         entity is deleted.

   228         """

   229         sqlcursor = session.pool[self.uri]        

   230         attrs = {'eid': eid}

   231         sql = self.sqladapter.sqlgen.delete(etype, attrs)

   232         sqlcursor.execute(sql, attrs)

   233     

   234     def delete_relation(self, session, subject, rtype, object):

   235         """delete a relation from the source"""

   236         rschema = self.schema.rschema(rtype)

   237         if rschema.inlined:

   238             if subject in session.query_data('pendingeids', ()):

   239                 return

   240             etype = session.describe(subject)[0]

   241             sql = 'UPDATE %s SET %s=NULL WHERE eid=%%(eid)s' % (etype, rtype)

   242             attrs = {'eid' : subject}

   243         else:

   244             attrs = {'eid_from': subject, 'eid_to': object}

   245             sql = self.sqladapter.sqlgen.delete('%s_relation' % rtype, attrs)

   246         sqlcursor = session.pool[self.uri]        

   247         sqlcursor.execute(sql, attrs)