本文整理汇总了Python中xapian.inmemory_open函数的典型用法代码示例。如果您正苦于以下问题:Python inmemory_open函数的具体用法?Python inmemory_open怎么用?Python inmemory_open使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了inmemory_open函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: add_from_purchased_but_needs_reinstall_data
def add_from_purchased_but_needs_reinstall_data(purchased_but_may_need_reinstall_list, db, cache):
"""Add application that have been purchased but may require a reinstall
This adds a inmemory database to the main db with the special
PURCHASED_NEEDS_REINSTALL_MAGIC_CHANNEL_NAME channel prefix
:return: a xapian query to get all the apps that need reinstall
"""
# magic
db_purchased = xapian.inmemory_open()
# go over the items we have
for item in purchased_but_may_need_reinstall_list:
# FIXME: what to do with duplicated entries? we will end
# up with two xapian.Document, one for the for-pay
# and one for the availalbe one from s-c-agent
#try:
# db.get_xapian_document(item.name,
# item.package_name)
#except IndexError:
# # item is not in the xapian db
# pass
#else:
# # ignore items we already have in the db, ignore
# continue
# index the item
try:
parser = SCAPurchasedApplicationParser(item)
index_app_info_from_parser(parser, db_purchased, cache)
except Exception as e:
LOG.exception("error processing: %s " % e)
# add new in memory db to the main db
db.add_database(db_purchased)
# return a query
query = xapian.Query("AH"+PURCHASED_NEEDS_REINSTALL_MAGIC_CHANNEL_NAME)
return query
开发者ID:cs2c,项目名称:AppStream,代码行数:35,代码来源:update.py
示例2: test_build_from_software_center_agent
def test_build_from_software_center_agent(self, mock_find_oauth):
# pretend we have no token
mock_find_oauth.return_value = None
db = xapian.inmemory_open()
cache = apt.Cache()
# monkey patch distro to ensure we get data
distro = softwarecenter.distro.get_distro()
distro.get_codename = lambda: "natty"
# we test against the real https://software-center.ubuntu.com here
# so we need network
res = update_from_software_center_agent(db, cache, ignore_cache=True)
# check results
self.assertTrue(res)
self.assertTrue(db.get_doccount() > 1)
for p in db.postlist(""):
doc = db.get_document(p.docid)
ppa = doc.get_value(XapianValues.ARCHIVE_PPA)
self.assertTrue(
ppa.startswith("commercial-ppa") and ppa.count("/") == 1, "ARCHIVE_PPA value incorrect, got '%s'" % ppa
)
self.assertTrue("-icon-" in doc.get_value(XapianValues.ICON))
# check support url in the DB
url = doc.get_value(XapianValues.SUPPORT_SITE_URL)
if url:
self.assertTrue(url.startswith("http") or url.startswith("mailto:"))
开发者ID:mortenpi,项目名称:ubuntu-software-center,代码行数:25,代码来源:test_database.py
示例3: test_reinstall_purchased_xapian
def test_reinstall_purchased_xapian(self, mock_helper, mock_agent):
small_available = [ self.available[0] ]
mock_agent.return_value = self._make_fake_scagent(
small_available, self.available_for_me)
db = xapian.inmemory_open()
cache = get_test_pkg_info()
# now create purchased debs xapian index (in memory because
# we store the repository passwords in here)
old_db_len = db.get_doccount()
update_from_software_center_agent(db, cache)
# ensure we have the new item
self.assertEqual(db.get_doccount(), old_db_len+2)
# query
query = get_reinstall_previous_purchases_query()
enquire = xapian.Enquire(db)
enquire.set_query(query)
matches = enquire.get_mset(0, db.get_doccount())
self.assertEqual(len(matches), 1)
distroseries = platform.dist()[2]
for m in matches:
doc = db.get_document(m.docid)
self.assertEqual(doc.get_value(XapianValues.PKGNAME), "photobomb")
self.assertEqual(
doc.get_value(XapianValues.ARCHIVE_SIGNING_KEY_ID),
"1024R/75254D99")
self.assertEqual(doc.get_value(XapianValues.ARCHIVE_DEB_LINE),
"deb https://username:[email protected]"
"private-ppa.launchpad.net/commercial-ppa-uploaders"
"/photobomb/ubuntu %s main" % distroseries)
开发者ID:pombredanne,项目名称:shop,代码行数:31,代码来源:test_reinstall_purchased.py
示例4: test_update_from_json_string
def test_update_from_json_string(self):
db = xapian.inmemory_open()
cache = apt.Cache()
p = os.path.join(DATA_DIR, "app-info-json", "apps.json")
res = update_from_json_string(db, cache, open(p).read(), origin=p)
self.assertTrue(res)
self.assertEqual(db.get_doccount(), 1)
开发者ID:mortenpi,项目名称:ubuntu-software-center,代码行数:7,代码来源:test_database.py
示例5: get_test_db_from_app_install_data
def get_test_db_from_app_install_data(datadir):
db = xapian.inmemory_open()
cache = get_pkg_info()
cache.open()
res = update_from_app_install_data(db, cache, datadir)
if res is False:
raise AssertionError("Failed to build db from '%s'" % datadir)
return db
开发者ID:pombredanne,项目名称:shop,代码行数:8,代码来源:utils.py
示例6: test_update_from_appstream_xml
def test_update_from_appstream_xml(self):
db = xapian.inmemory_open()
res = update_from_appstream_xml(db, self.cache, os.path.join(DATA_DIR, "app-info"))
self.assertTrue(res)
self.assertEqual(db.get_doccount(), 1)
# FIXME: improve tests
for p in db.postlist(""):
doc = db.get_document(p.docid)
for term in doc.termlist():
self.assertIsInstance(term, xapian.TermListItem)
self.assertIsInstance(term.term, basestring)
for value in doc.values():
self.assertIsInstance(value, xapian.ValueItem)
self.assertIsInstance(value.num, long)
self.assertIsInstance(value.value, basestring)
开发者ID:mortenpi,项目名称:ubuntu-software-center,代码行数:15,代码来源:test_database.py
示例7: test_for_purchase_apps_date_published
def test_for_purchase_apps_date_published(self, mock_find_oauth):
# pretend we have no token
mock_find_oauth.return_value = None
# os.environ["SOFTWARE_CENTER_DEBUG_HTTP"] = "1"
# os.environ["SOFTWARE_CENTER_AGENT_HOST"] = "http://sc.staging.ubuntu.com/"
# staging does not have a valid cert
os.environ["PISTON_MINI_CLIENT_DISABLE_SSL_VALIDATION"] = "1"
cache = get_test_pkg_info()
db = xapian.inmemory_open()
res = update_from_software_center_agent(db, cache, ignore_cache=True)
self.assertTrue(res)
for p in db.postlist(""):
doc = db.get_document(p.docid)
date_published = doc.get_value(XapianValues.DATE_PUBLISHED)
# make sure that a date_published value is provided
self.assertNotEqual(date_published, "")
self.assertNotEqual(date_published, None)
开发者ID:mortenpi,项目名称:ubuntu-software-center,代码行数:18,代码来源:test_database.py
示例8: make_catalog
def make_catalog(uri, fields):
"""Creates a new and empty catalog in the given uri.
If uri=None the catalog is made "in memory".
fields must be a dict. It contains some informations about the
fields in the database.
By example:
fields = {'id': Integer(is_key_field=True, is_stored=True,
is_indexed=True), ...}
"""
# In memory
if uri is None:
db = inmemory_open()
return Catalog(db, fields, asynchronous_mode=False)
# In the local filesystem
path = lfs.get_absolute_path(uri)
db = WritableDatabase(path, DB_CREATE)
return Catalog(db, fields)
开发者ID:kennym,项目名称:itools,代码行数:19,代码来源:catalog.py
示例9: add_from_purchased_but_needs_reinstall_data
def add_from_purchased_but_needs_reinstall_data(purchased_but_may_need_reinstall_list, db, cache):
"""Add application that have been purchased but may require a reinstall
This adds a inmemory database to the main db with the special
PURCHASED_NEEDS_REINSTALL_MAGIC_CHANNEL_NAME channel prefix
:return: a xapian query to get all the apps that need reinstall
"""
# magic
db_purchased = xapian.inmemory_open()
# go over the items we have
for item in purchased_but_may_need_reinstall_list:
# FIXME: what to do with duplicated entries? we will end
# up with two xapian.Document, one for the for-pay
# and one for the availalbe one from s-c-agent
#try:
# db.get_xapian_document(item.name,
# item.package_name)
#except IndexError:
# # item is not in the xapian db
# pass
#else:
# # ignore items we already have in the db, ignore
# continue
# index the item
try:
# we fake a channel here
item.channel = PURCHASED_NEEDS_REINSTALL_MAGIC_CHANNEL_NAME
# and empty category to make the parser happy
item.categories = ""
# WARNING: item.name needs to be different than
# the item.name in the DB otherwise the DB
# gets confused about (appname, pkgname) duplication
item.name = utf8(_("%s (already purchased)")) % utf8(item.name)
parser = SoftwareCenterAgentParser(item)
index_app_info_from_parser(parser, db_purchased, cache)
except Exception as e:
LOG.exception("error processing: %s " % e)
# add new in memory db to the main db
db.add_database(db_purchased)
# return a query
query = xapian.Query("AH"+PURCHASED_NEEDS_REINSTALL_MAGIC_CHANNEL_NAME)
return query
开发者ID:armikhael,项目名称:software-center,代码行数:43,代码来源:update.py
示例10: test_preserve_enquire_sorter
def test_preserve_enquire_sorter():
"""Test preservation of sorter set on enquire.
"""
db = xapian.inmemory_open()
doc = xapian.Document()
doc.add_term("foo")
doc.add_value(1, "1")
db.add_document(doc)
db.add_document(doc)
def make_enq1(db):
enq = xapian.Enquire(db)
sorter = xapian.MultiValueSorter()
enq.set_sort_by_key(sorter, True)
del sorter
return enq
enq = make_enq1(db)
enq.set_query(xapian.Query("foo"))
enq.get_mset(0, 10)
def make_enq2(db):
enq = xapian.Enquire(db)
sorter = xapian.MultiValueSorter()
enq.set_sort_by_key_then_relevance(sorter, True)
del sorter
return enq
enq = make_enq2(db)
enq.set_query(xapian.Query("foo"))
enq.get_mset(0, 10)
def make_enq3(db):
enq = xapian.Enquire(db)
sorter = xapian.MultiValueSorter()
enq.set_sort_by_relevance_then_key(sorter, True)
del sorter
return enq
enq = make_enq3(db)
enq.set_query(xapian.Query("foo"))
enq.get_mset(0, 10)
开发者ID:wangeguo,项目名称:xapian,代码行数:43,代码来源:pythontest2.py
示例11: _database
def _database(self, writable=False):
"""
Private method that returns a xapian.Database for use.
Optional arguments:
``writable`` -- Open the database in read/write mode (default=False)
Returns an instance of a xapian.Database or xapian.WritableDatabase
"""
if settings.HAYSTACK_XAPIAN_PATH == MEMORY_DB_NAME:
if not SearchBackend.inmemory_db:
SearchBackend.inmemory_db = xapian.inmemory_open()
return SearchBackend.inmemory_db
if writable:
database = xapian.WritableDatabase(settings.HAYSTACK_XAPIAN_PATH, xapian.DB_CREATE_OR_OPEN)
else:
try:
database = xapian.Database(settings.HAYSTACK_XAPIAN_PATH)
except xapian.DatabaseOpeningError:
raise InvalidIndexError(u"Unable to open index at %s" % settings.HAYSTACK_XAPIAN_PATH)
return database
开发者ID:domasx2,项目名称:xapian-haystack,代码行数:22,代码来源:xapian_backend.py
示例12: _database
def _database(self, writable=False):
"""
Private method that returns a xapian.Database for use.
Optional arguments:
``writable`` -- Open the database in read/write mode (default=False)
Returns an instance of a xapian.Database or xapian.WritableDatabase
"""
if self.path == MEMORY_DB_NAME:
if not self.inmemory_db:
self.inmemory_db = xapian.inmemory_open()
return self.inmemory_db
if writable:
database = xapian.WritableDatabase(self.path, xapian.DB_CREATE_OR_OPEN)
else:
try:
database = xapian.Database(self.path)
except xapian.DatabaseOpeningError:
raise InvalidIndexError(u'Unable to open index at %s' % self.path)
return database
开发者ID:Findeton,项目名称:agora-ciudadana,代码行数:22,代码来源:xapian_backend.py
示例13: _database
def _database(self, writable=False):
"""
Private method that returns a xapian.Database for use.
Optional arguments:
``writable`` -- Open the database in read/write mode (default=False)
Returns an instance of a xapian.Database or xapian.WritableDatabase
"""
if isinstance(settings.HAYSTACK_XAPIAN_PATH, basestring):
if self.path == MEMORY_DB_NAME:
if not self.inmemory_db:
self.inmemory_db = xapian.inmemory_open()
return self.inmemory_db
if writable:
database = xapian.WritableDatabase(self.path, xapian.DB_CREATE_OR_OPEN)
else:
try:
database = xapian.Database(self.path)
except xapian.DatabaseOpeningError:
raise InvalidIndexError(u'Unable to open index at %s' % self.path)
else:
try:
host, port = settings.HAYSTACK_XAPIAN_PATH
except ValueError:
raise InvalidIndexError(u'Unable to open index at %s' % settings.HAYSTACK_XAPIAN_PATH)
if writable:
database = xapian.remote_open_writable(host, port)
database = xapian.WritableDatabase(database)
else:
try:
database = xapian.remote_open(host, port)
database = xapian.Database(database)
except xapian.DatabaseOpeningError:
raise InvalidIndexError(u'Unable to open index at %s' % settings.HAYSTACK_XAPIAN_PATH)
return database
开发者ID:aarcro,项目名称:xapian-haystack,代码行数:39,代码来源:xapian_backend.py
示例14: setup_database
def setup_database():
"""Set up and return an inmemory database with 5 documents.
"""
db = xapian.inmemory_open()
doc = xapian.Document()
doc.set_data("is it cold?")
doc.add_term("is")
doc.add_posting("it", 1)
doc.add_posting("cold", 2)
db.add_document(doc)
doc = xapian.Document()
doc.set_data("was it warm?")
doc.add_posting("was", 1)
doc.add_posting("it", 2)
doc.add_posting("warm", 3)
db.add_document(doc)
doc.set_data("was it warm? two")
doc.add_term("two", 2)
doc.add_value(0, xapian.sortable_serialise(2))
db.add_document(doc)
doc.set_data("was it warm? three")
doc.add_term("three", 3)
doc.add_value(0, xapian.sortable_serialise(1.5))
db.add_document(doc)
doc.set_data("was it warm? four it")
doc.add_term("four", 4)
doc.add_term("it", 6)
doc.add_posting("it", 7)
doc.add_value(5, "five")
doc.add_value(9, "nine")
doc.add_value(0, xapian.sortable_serialise(2))
db.add_document(doc)
expect(db.get_doccount(), 5)
return db
开发者ID:wangeguo,项目名称:xapian,代码行数:38,代码来源:pythontest2.py
示例15: test_update_from_var_lib_apt_lists
def test_update_from_var_lib_apt_lists(self):
# ensure we index with german locales to test i18n
os.environ["LANGUAGE"] = "de"
db = xapian.inmemory_open()
res = update_from_var_lib_apt_lists(db, self.cache, listsdir=os.path.join(DATA_DIR, "app-info"))
self.assertTrue(res)
self.assertEqual(db.get_doccount(), 1)
# test if Name-de was picked up
i = 0
for it in db.postlist("AAFestplattenbelegung analysieren"):
i += 1
self.assertEqual(i, 1)
# test if gettext worked
found_gettext_translation = False
for it in db.postlist("AAFestplattenbelegung analysieren"):
doc = db.get_document(it.docid)
for term_iter in doc.termlist():
# a german term from the app-info file to ensure that
# it got indexed in german
if term_iter.term == "festplattenbelegung":
found_gettext_translation = True
break
self.assertTrue(found_gettext_translation)
开发者ID:mortenpi,项目名称:ubuntu-software-center,代码行数:23,代码来源:test_database.py
示例16: test_all
def test_all():
# Test the version number reporting functions give plausible results.
v = "%d.%d.%d" % (xapian.major_version(),
xapian.minor_version(),
xapian.revision())
v2 = xapian.version_string()
expect(v2, v, "Unexpected version output")
# A regexp check would be better, but seems to create a bogus "leak" of -1
# objects in Python 3.
expect(len(xapian.__version__.split('.')), 3, 'xapian.__version__ not X.Y.Z')
expect((xapian.__version__.split('.'))[0], '1', 'xapian.__version__ not "1.Y.Z"')
def access_cvar():
res = xapian.cvar
print("Unhandled constants: ", res)
return res
# Check that SWIG isn't generating cvar (regression test for ticket#297).
expect_exception(AttributeError, "'module' object has no attribute 'cvar'",
access_cvar)
stem = xapian.Stem(b"english")
expect(str(stem), "Xapian::Stem(english)", "Unexpected str(stem)")
doc = xapian.Document()
doc.set_data(b"a\0b")
if doc.get_data() == b"a":
raise TestFail("get_data+set_data truncates at a zero byte")
expect(doc.get_data(), b"a\0b", "get_data+set_data doesn't transparently handle a zero byte")
doc.set_data(b"is there anybody out there?")
doc.add_term(b"XYzzy")
doc.add_posting(stem(b"is"), 1)
doc.add_posting(stem(b"there"), 2)
doc.add_posting(stem(b"anybody"), 3)
doc.add_posting(stem(b"out"), 4)
doc.add_posting(stem(b"there"), 5)
db = xapian.inmemory_open()
db.add_document(doc)
expect(db.get_doccount(), 1, "Unexpected db.get_doccount()")
terms = ["smoke", "test", "terms"]
expect_query(xapian.Query(xapian.Query.OP_OR, [t.encode('utf-8') for t in terms]),
"(smoke OR test OR terms)")
query1 = xapian.Query(xapian.Query.OP_PHRASE, (b"smoke", b"test", b"tuple"))
query2 = xapian.Query(xapian.Query.OP_XOR, (xapian.Query(b"smoke"), query1, b"string"))
expect_query(query1, "(smoke PHRASE 3 test PHRASE 3 tuple)")
expect_query(query2, "(smoke XOR (smoke PHRASE 3 test PHRASE 3 tuple) XOR string)")
subqs = ["a", "b"]
expect_query(xapian.Query(xapian.Query.OP_OR, [s.encode('utf-8') for s in subqs]), "(a OR b)")
expect_query(xapian.Query(xapian.Query.OP_VALUE_RANGE, 0, b'1', b'4'),
"VALUE_RANGE 0 1 4")
# Check database factory functions are wrapped as expected (or not wrapped
# in the first cases):
expect_exception(AttributeError, "'module' object has no attribute 'open_stub'",
lambda : xapian.open_stub(b"nosuchdir/nosuchdb"))
expect_exception(AttributeError, "'module' object has no attribute 'open_stub'",
lambda : xapian.open_stub(b"nosuchdir/nosuchdb", xapian.DB_OPEN))
expect_exception(AttributeError, "'module' object has no attribute 'chert_open'",
lambda : xapian.chert_open(b"nosuchdir/nosuchdb"))
expect_exception(AttributeError, "'module' object has no attribute 'chert_open'",
lambda : xapian.chert_open(b"nosuchdir/nosuchdb", xapian.DB_CREATE))
expect_exception(xapian.DatabaseOpeningError, None,
lambda : xapian.Database(b"nosuchdir/nosuchdb", xapian.DB_BACKEND_STUB))
expect_exception(xapian.DatabaseOpeningError, None,
lambda : xapian.WritableDatabase(b"nosuchdir/nosuchdb", xapian.DB_OPEN|xapian.DB_BACKEND_STUB))
expect_exception(xapian.DatabaseOpeningError, None,
lambda : xapian.Database(b"nosuchdir/nosuchdb", xapian.DB_BACKEND_GLASS))
expect_exception(xapian.DatabaseCreateError, None,
lambda : xapian.WritableDatabase(b"nosuchdir/nosuchdb", xapian.DB_CREATE|xapian.DB_BACKEND_GLASS))
expect_exception(xapian.DatabaseOpeningError, None,
lambda : xapian.Database(b"nosuchdir/nosuchdb", xapian.DB_BACKEND_CHERT))
expect_exception(xapian.DatabaseCreateError, None,
lambda : xapian.WritableDatabase(b"nosuchdir/nosuchdb", xapian.DB_CREATE|xapian.DB_BACKEND_CHERT))
expect_exception(xapian.NetworkError, None,
xapian.remote_open, b"/bin/false", b"")
expect_exception(xapian.NetworkError, None,
xapian.remote_open_writable, b"/bin/false", b"")
expect_exception(xapian.NetworkError, None,
xapian.remote_open, b"127.0.0.1", 0, 1)
expect_exception(xapian.NetworkError, None,
xapian.remote_open_writable, b"127.0.0.1", 0, 1)
# Check wrapping of MatchAll and MatchNothing:
expect_query(xapian.Query.MatchAll, "<alldocuments>")
expect_query(xapian.Query.MatchNothing, "")
# Feature test for Query.__iter__
term_count = 0
for term in query2:
term_count += 1
#.........这里部分代码省略.........
开发者ID:PriyankBhatt,项目名称:xapian,代码行数:101,代码来源:smoketest.py
示例17: test_all
def test_all():
# Test the version number reporting functions give plausible results.
v = "%d.%d.%d" % (xapian.major_version(), xapian.minor_version(), xapian.revision())
v2 = xapian.version_string()
expect(v2, v, "Unexpected version output")
def access_cvar():
return xapian.cvar
# Check that SWIG isn't generating cvar (regression test for ticket#297).
expect_exception(AttributeError, "'module' object has no attribute 'cvar'", access_cvar)
stem = xapian.Stem("english")
expect(str(stem), "Xapian::Stem(english)", "Unexpected str(stem)")
doc = xapian.Document()
doc.set_data("a\0b")
if doc.get_data() == "a":
raise TestFail("get_data+set_data truncates at a zero byte")
expect(doc.get_data(), "a\0b", "get_data+set_data doesn't transparently handle a zero byte")
doc.set_data("is there anybody out there?")
doc.add_term("XYzzy")
doc.add_posting(stem("is"), 1)
doc.add_posting(stem("there"), 2)
doc.add_posting(stem("anybody"), 3)
doc.add_posting(stem("out"), 4)
doc.add_posting(stem("there"), 5)
db = xapian.inmemory_open()
db.add_document(doc)
expect(db.get_doccount(), 1, "Unexpected db.get_doccount()")
terms = ["smoke", "test", "terms"]
expect_query(xapian.Query(xapian.Query.OP_OR, terms), "(smoke OR test OR terms)")
query1 = xapian.Query(xapian.Query.OP_PHRASE, ("smoke", "test", "tuple"))
query2 = xapian.Query(xapian.Query.OP_XOR, (xapian.Query("smoke"), query1, "string"))
expect_query(query1, "(smoke PHRASE 3 test PHRASE 3 tuple)")
expect_query(query2, "(smoke XOR (smoke PHRASE 3 test PHRASE 3 tuple) XOR string)")
subqs = ["a", "b"]
expect_query(xapian.Query(xapian.Query.OP_OR, subqs), "(a OR b)")
expect_query(xapian.Query(xapian.Query.OP_VALUE_RANGE, 0, "1", "4"), "VALUE_RANGE 0 1 4")
# Check database factory functions are wrapped as expected:
expect_exception(xapian.DatabaseOpeningError, None, xapian.open_stub, "nosuchdir/nosuchdb")
expect_exception(xapian.DatabaseOpeningError, None, xapian.open_stub, "nosuchdir/nosuchdb", xapian.DB_OPEN)
expect_exception(xapian.DatabaseOpeningError, None, xapian.brass_open, "nosuchdir/nosuchdb")
expect_exception(xapian.DatabaseCreateError, None, xapian.brass_open, "nosuchdir/nosuchdb", xapian.DB_CREATE)
expect_exception(xapian.DatabaseOpeningError, None, xapian.chert_open, "nosuchdir/nosuchdb")
expect_exception(xapian.DatabaseCreateError, None, xapian.chert_open, "nosuchdir/nosuchdb", xapian.DB_CREATE)
expect_exception(xapian.NetworkError, None, xapian.remote_open, "/bin/false", "")
expect_exception(xapian.NetworkError, None, xapian.remote_open_writable, "/bin/false", "")
expect_exception(xapian.NetworkError, None, xapian.remote_open, "127.0.0.1", 0, 1)
expect_exception(xapian.NetworkError, None, xapian.remote_open_writable, "127.0.0.1", 0, 1)
# Check wrapping of MatchAll and MatchNothing:
expect_query(xapian.Query.MatchAll, "<alldocuments>")
expect_query(xapian.Query.MatchNothing, "")
# Feature test for Query.__iter__
term_count = 0
for term in query2:
term_count += 1
expect(term_count, 4, "Unexpected number of terms in query2")
enq = xapian.Enquire(db)
enq.set_query(xapian.Query(xapian.Query.OP_OR, "there", "is"))
mset = enq.get_mset(0, 10)
expect(mset.size(), 1, "Unexpected mset.size()")
expect(len(mset), 1, "Unexpected mset.size()")
# Feature test for Enquire.matching_terms(docid)
term_count = 0
for term in enq.matching_terms(mset.get_hit(0)):
term_count += 1
expect(term_count, 2, "Unexpected number of matching terms")
# Feature test for MSet.__iter__
msize = 0
for match in mset:
msize += 1
expect(msize, mset.size(), "Unexpected number of entries in mset")
terms = " ".join(enq.matching_terms(mset.get_hit(0)))
expect(terms, "is there", "Unexpected terms")
# Feature test for ESet.__iter__
rset = xapian.RSet()
rset.add_document(1)
eset = enq.get_eset(10, rset)
term_count = 0
for term in eset:
term_count += 1
expect(term_count, 3, "Unexpected number of expand terms")
# Feature test for Database.__iter__
#.........这里部分代码省略.........
开发者ID:RileyRC,项目名称:xapian,代码行数:101,代码来源:smoketest2.py
示例18: inmemory
def inmemory():
"""Returns an xodb database backed by an in-memory xapian
database. Does not support spelling correction.
"""
return open(xapian.inmemory_open(), spelling=False, inmem=True)
开发者ID:7footmoustache,项目名称:xodb,代码行数:5,代码来源:__init__.py
示例19: time
if __name__ == "__main__":
import sys
from time import time
import linecache
import glob
import traceback
import linecache
import xapian
import re
stem = xapian.Stem("french")
ti = xapian.inmemory_open()
ti = xapian.WritableDatabase("test.ti", xapian.DB_CREATE_OR_OPEN)
# ti = xapian.quartz_open('test.idx')
# start = time()
# lines = 0
# for f in glob.glob('*.txt'):
# print f,
# for linenumber, line in enumerate(file(f,'rb')):
# lines += 1
# line = line.strip()
# doc = xapian.Document()
# doc.set_data('%12s:%04i'%(f,linenumber))
# for word_number, word in enumerate(re.findall(r'\w+',line.lower())):
# doc.add_posting(word,word_number)
# ti.add_document(doc)
# if linenumber % 100 == 0:
# sys.stdout.write('.')
# print 'OK'
# print 'Indexing time : %.2fs for %i lines'%(time()-start,lines)
开发者ID:atbrox,项目名称:pytst,代码行数:31,代码来源:testxapian.py
示例20:
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import xapian
stem = xapian.Stem('english')
db = xapian.inmemory_open()
doc = xapian.Document()
doc.add_posting(stem("is"), 1)
doc.add_posting(stem("there"), 2)
doc.add_posting(stem("anybody"), 3)
doc.add_posting(stem("out"), 4)
doc.add_posting(stem("there"), 5)
db.add_document(doc)
doc1 = xapian.Document()
doc1.add_posting(stem("is"), 1)
doc1.add_posting(stem("there"), 2)
doc1.add_posting(stem("anybody"), 3)
doc1.add_posting(stem("out"), 4)
doc1.add_posting(stem("there"), 5)
db.add_document(doc1)
db.commit()
for term in db.allterms():
print term.term, term.termfreq
"""
anybodi 2
is 2
开发者ID:FashtimeDotCom,项目名称:xapian_weibo,代码行数:31,代码来源:test_allterm.py
注:本文中的xapian.inmemory_open函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论