()
| 24 | |
| 25 | |
| 26 | async def run(): |
| 27 | my_cluster = Cluster("couchbase://localhost", opts) |
| 28 | await my_cluster.on_connect() |
| 29 | |
| 30 | bucket = my_cluster.bucket("default") |
| 31 | await bucket.on_connect() |
| 32 | |
| 33 | coll = bucket.default_collection() |
| 34 | |
| 35 | key1 = str(uuid4()) |
| 36 | key2 = str(uuid4()) |
| 37 | |
| 38 | tasks = [coll.insert(key1, {"value": 10}), coll.insert(key2, {"value": 0})] |
| 39 | results = await wait_for_all(tasks) |
| 40 | print(f'insert results: {results}') |
| 41 | |
| 42 | async def txn_logic(ctx # type: AttemptContext |
| 43 | ): |
| 44 | t = [ctx.get(coll, key1), ctx.get(coll, key2)] |
| 45 | res = await wait_for_all(t) |
| 46 | print(f'get results: {results}') |
| 47 | doc1 = res[0] |
| 48 | doc2 = res[1] |
| 49 | doc1_content = doc1.content_as[dict] |
| 50 | doc2_content = doc2.content_as[dict] |
| 51 | print(f'doc1:{doc1}, doc2: {doc2}') |
| 52 | print(f'doc1_content: {doc1_content}, doc2_content: {doc2_content}') |
| 53 | if doc1_content["value"] > 0: |
| 54 | doc1_content["value"] = doc1_content["value"] - 1 |
| 55 | doc2_content["value"] = doc2_content["value"] + 1 |
| 56 | t = [ctx.replace(doc1, doc1_content), ctx.replace(doc2, doc2_content)] |
| 57 | await wait_for_all(t) |
| 58 | else: |
| 59 | raise RuntimeError("doc1 is exhausted") |
| 60 | |
| 61 | ok = True |
| 62 | while ok: |
| 63 | try: |
| 64 | print(f'txn_result: {await my_cluster.transactions.run(txn_logic)}') |
| 65 | except TransactionFailed as e: |
| 66 | print(f'txn raised exception: {e}') |
| 67 | ok = False |
| 68 | |
| 69 | tasks = [coll.get(key1), coll.get(key2)] |
| 70 | results = await wait_for_all(tasks) |
| 71 | print(f'after txns, we have {results}') |
| 72 | |
| 73 | |
| 74 | loop = asyncio.get_event_loop() |
no test coverage detected