()
| 135 | |
| 136 | |
| 137 | async def main(): |
| 138 | auth = PasswordAuthenticator('Administrator', 'password') |
| 139 | opts = ClusterOptions(auth) |
| 140 | cluster = await AsyncCluster.connect('couchbase://localhost', opts) |
| 141 | bucket = cluster.bucket('default') |
| 142 | await bucket.on_connect() |
| 143 | collection = bucket.default_collection() |
| 144 | |
| 145 | # NOTE: validate paths (see note on line 18) if going to use setup/teardown functionality |
| 146 | keys = await setup(cluster, collection) |
| 147 | vector = load_test_vector() |
| 148 | |
| 149 | search_req = search.SearchRequest.create(search.MatchAllQuery()).with_vector_search( |
| 150 | VectorSearch.from_vector_query(VectorQuery('vector_field', vector))) |
| 151 | search_iter = cluster.search(TEST_INDEX_NAME, search_req, SearchOptions(limit=2)) |
| 152 | async for row in search_iter.rows(): |
| 153 | print(f'row: {row}') |
| 154 | |
| 155 | print(f'Metatdata: {search_iter.metadata()}') |
| 156 | |
| 157 | # NOTE: only use in conjunction w/ setup() method |
| 158 | await teardown(cluster, collection, keys) |
| 159 | |
| 160 | |
| 161 | if __name__ == '__main__': |
no test coverage detected