cubicweb: comparison server/sources/native.py

equal deleted inserted replaced

-:c77629112437
+:9eadf34fd860
 NONSYSTEM_RELATIONS = set()
 class LogCursor(object):
 def __init__(self, cursor):
 self.cu = cursor
 def execute(self, query, args=None):
 """Execute a query.
 it's a function just so that it shows up in profiling
 """
 if server.DEBUG:
 self.cu.execute(str(query), args)
 except Exception, ex:
 print "sql: %r\n args: %s\ndbms message: %r" % (
 query, args, ex.args[0])
 raise
 def fetchall(self):
 return self.cu.fetchall()
 def fetchone(self):
 return self.cu.fetchone()
 def make_schema(selected, solution, table, typemap):
 """return a sql schema to store RQL query result"""
 sql = []
 varmap = {}
 for i, term in enumerate(selected):
 class NativeSQLSource(SQLAdapterMixIn, AbstractSource):
 """adapter for source using the native cubicweb schema (see below)
 """
 # need default value on class since migration doesn't call init method
 has_deleted_entitites_table = True
 passwd_rql = "Any P WHERE X is CWUser, X login %(login)s, X upassword P"
 auth_rql = "Any X WHERE X is CWUser, X login %(login)s, X upassword %(pwd)s"
 _sols = ({'X': 'CWUser', 'P': 'Password'},)
 options = (
 ('db-driver',
 {'type' : 'string',
 'default': 'postgres',
 'help': 'database driver (postgres or sqlite)',
 'default': 'utf8',
 'help': 'database encoding',
 'group': 'native-source', 'inputlevel': 1,
 }),
 )
 def __init__(self, repo, appschema, source_config, *args, **kwargs):
 SQLAdapterMixIn.__init__(self, source_config)
 AbstractSource.__init__(self, repo, appschema, source_config,
 *args, **kwargs)
 # sql generator
 self._eid_creation_lock = Lock()
 def reset_caches(self):
 """method called during test to reset potential source caches"""
 self._cache = Cache(self.repo.config['rql-cache-size'])
 def clear_eid_cache(self, eid, etype):
 """clear potential caches for the given eid"""
 self._cache.pop('%s X WHERE X eid %s' % (etype, eid), None)
 self._cache.pop('Any X WHERE X eid %s' % eid, None)
 def sqlexec(self, session, sql, args=None):
 """execute the query and return its result"""
 cursor = session.pool[self.uri]
 self.doexec(cursor, sql, args)
 return self.process_result(cursor)
 def init_creating(self):
 # check full text index availibility
 pool = self.repo._get_pool()
 if not self.indexer.has_fti_table(pool['system']):
 self.error('no text index table')
 self.indexer = None
 self.repo._free_pool(pool)
 def init(self):
 self.init_creating()
 pool = self.repo._get_pool()
 # XXX cubicweb < 2.42 compat
 if 'deleted_entities' in self.dbhelper.list_tables(pool['system']):
 self.has_deleted_entitites_table = True
 else:
 self.has_deleted_entitites_table = False
 self.repo._free_pool(pool)
 # ISource interface #######################################################
 def compile_rql(self, rql):
 rqlst = self.repo.querier._rqlhelper.parse(rql)
 rqlst.restricted_vars = ()
 rqlst.children[0].solutions = self._sols
 self.repo.querier.sqlgen_annotate(rqlst)
 set_qdata(self.schema.rschema, rqlst, ())
 return rqlst
 def set_schema(self, schema):
 """set the application'schema"""
 self._cache = Cache(self.repo.config['rql-cache-size'])
 self.cache_hit, self.cache_miss, self.no_cache = 0, 0, 0
 self.schema = schema
 pass # __init__
 if 'CWUser' in schema: # probably an empty schema if not true...
 # rql syntax trees used to authenticate users
 self._passwd_rqlst = self.compile_rql(self.passwd_rql)
 self._auth_rqlst = self.compile_rql(self.auth_rql)
 def support_entity(self, etype, write=False):
 """return true if the given entity's type is handled by this adapter
 if write is true, return true only if it's a RW support
 """
 return not etype in NONSYSTEM_ETYPES
 def support_relation(self, rtype, write=False):
 """return true if the given relation's type is handled by this adapter
 if write is true, return true only if it's a RW support
 """
 if write:
 return not rtype in NONSYSTEM_RELATIONS
 # due to current multi-sources implementation, the system source
 # can't claim not supporting a relation
 return True #not rtype == 'content_for'
 def authenticate(self, session, login, password):
 """return CWUser eid for the given login/password if this account is
 defined in this source, else raise `AuthenticationError`
 rset = self.syntax_tree_search(session, self._auth_rqlst, args)
 try:
 return rset[0][0]
 except IndexError:
 raise AuthenticationError('bad password')
 def syntax_tree_search(self, session, union, args=None, cachekey=None,
 varmap=None):
 """return result from this source for a rql query (actually from
 a rql syntax tree and a solution dictionary mapping each used
 variable to a possible type). If cachekey is given, the query
 necessary to fetch the results (but not the results themselves)
 self.doexec(cursor, sql, args)
 res = self.process_result(cursor)
 if server.DEBUG:
 print '------>', res
 return res
 def flying_insert(self, table, session, union, args=None, varmap=None):
 """similar as .syntax_tree_search, but inserts data in the
 temporary table (on-the-fly if possible, eg for the system
 source whose the given cursor come from). If not possible,
 inserts all data by calling .executemany().
 #                 print 'data', row
 #                 tempdata.add(tuple(row))
 else:
 super(NativeSQLSource, self).flying_insert(table, session, union,
 args, varmap)
 def _manual_insert(self, results, table, session):
 """insert given result into a temporary table on the system source"""
 #print 'manual insert', table, results
 if not results:
 return
 pass
 try:
 del self._temp_table_data[table]
 except KeyError:
 continue
 def add_entity(self, session, entity):
 """add a new entity to the source"""
 attrs = self.preprocess_entity(entity)
 sql = self.sqlgen.insert(SQL_PREFIX + str(entity.e_schema), attrs)
 self.doexec(session.pool[self.uri], sql, attrs)
 def update_entity(self, session, entity):
 """replace an entity in the source"""
 attrs = self.preprocess_entity(entity)
 sql = self.sqlgen.update(SQL_PREFIX + str(entity.e_schema), attrs, [SQL_PREFIX + 'eid'])
 self.doexec(session.pool[self.uri], sql, attrs)
 def add_relation(self, session, subject, rtype, object):
 """add a relation to the source"""
 attrs = {'eid_from': subject, 'eid_to': object}
 sql = self.sqlgen.insert('%s_relation' % rtype, attrs)
 self.doexec(session.pool[self.uri], sql, attrs)
 def delete_relation(self, session, subject, rtype, object):
 """delete a relation from the source"""
 rschema = self.schema.rschema(rtype)
 if rschema.inlined:
 table = SQL_PREFIX + session.describe(subject)[0]
 SQL_PREFIX)
 attrs = {'eid' : subject}
 else:
 attrs = {'eid_from': subject, 'eid_to': object}
 sql = self.sqlgen.delete('%s_relation' % rtype, attrs)
 self.doexec(session.pool[self.uri], sql, attrs)
 def doexec(self, cursor, query, args=None):
 """Execute a query.
 it's a function just so that it shows up in profiling
 """
 cursor.execute(str(query), args)
 except Exception, ex:
 self.critical("sql: %r\n args: %s\ndbms message: %r",
 query, args, ex.args[0])
 raise
 def doexecmany(self, cursor, query, args):
 """Execute a query.
 it's a function just so that it shows up in profiling
 """
 #t1 = time()
 try:
 cursor.executemany(str(query), args)
 except:
 self.critical("sql many: %r\n args: %s", query, args)
 raise
 # short cut to method requiring advanced db helper usage ##################
 def create_index(self, session, table, column, unique=False):
 cursor = LogCursor(session.pool[self.uri])
 self.dbhelper.create_index(cursor, table, column, unique)
 def drop_index(self, session, table, column, unique=False):
 cursor = LogCursor(session.pool[self.uri])
 self.dbhelper.drop_index(cursor, table, column, unique)
 # system source interface #################################################
 'extid=%(x)s AND source=%(s)s',
 # str() necessary with pg 8.3
 {'x': str(lid), 's': source.uri})
 # XXX testing rowcount cause strange bug with sqlite, results are there
 #     but rowcount is 0
 #if cursor.rowcount > 0:
 try:
 result = cursor.fetchone()
 if result:
 eid = result[0]
 return eid
 except:
 pass
 return None
 def temp_table_def(self, selected, sol, table):
 return make_schema(selected, sol, table, self.dbhelper.TYPE_MAPPING)
 def create_temp_table(self, session, table, schema):
 # we don't want on commit drop, this may cause problem when
 # running with an ldap source, and table will be deleted manually any way
 # on commit
 sql = self.dbhelper.sql_temporary_table(table, schema, False)
 self.doexec(session.pool[self.uri], sql)
 def create_eid(self, session):
 self._eid_creation_lock.acquire()
 try:
 cursor = session.pool[self.uri]
 for sql in self.dbhelper.sqls_increment_sequence('entities_id_seq'):
 session.system_sql(self.sqlgen.delete('entities', attrs), attrs)
 if self.has_deleted_entitites_table:
 attrs = {'type': etype, 'eid': eid, 'extid': extid,
 'source': uri, 'dtime': datetime.now()}
 session.system_sql(self.sqlgen.insert('deleted_entities', attrs), attrs)
 def fti_unindex_entity(self, session, eid):
 """remove text content for entity with the given eid from the full text
 index
 """
 try:
 self.indexer.cursor_unindex_object(eid, session.pool['system'])
 except:
 if self.indexer is not None:
 self.exception('error while unindexing %s', eid)
 def fti_index_entity(self, session, entity):
 """add text content of a created/modified entity to the full text index
 """
 self.info('reindexing %r', entity.eid)
 try:
 if self.indexer is not None:
 self.exception('error while reindexing %s', entity)
 # update entities.mtime
 attrs = {'eid': entity.eid, 'mtime': datetime.now()}
 session.system_sql(self.sqlgen.update('entities', attrs, ['eid']), attrs)
 def modified_entities(self, session, etypes, mtime):
 """return a 2-uple:
 * list of (etype, eid) of entities of the given types which have been
 modified since the given timestamp (actually entities whose full text
 index content has changed)

branch	tls-sprint
changeset 1792	9eadf34fd860
parent 1398	5fe84a5f7035
child 1952	8e19c813750d