Skip to content

Instantly share code, notes, and snippets.

@oberstet
Created March 22, 2026 17:29
Show Gist options
  • Select an option

  • Save oberstet/5ed283695209d5d78ad5de7a9425cd79 to your computer and use it in GitHub Desktop.

Select an option

Save oberstet/5ed283695209d5d78ad5de7a9425cd79 to your computer and use it in GitHub Desktop.
oberstet@amd-ryzen5:~/scm/typedefint/wampire$ find ~/work/wamp/crossbar/src/ -name "*.py" -exec grep -H "catalog" {} \;
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api09_catalog.py: # Get catalogs of the current user
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api09_catalog.py: result = await self.call("xbr.network.get_catalogs_by_owner", member_id.bytes)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api09_catalog.py: self.log.info(f"get_catalogs_by_owner results=\n\n{pformat(result)}\n")
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api09_catalog.py: # Get all catalogs
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api09_catalog.py: result = await self.call("xbr.network.find_catalogs")
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api09_catalog.py: self.log.info(f"find_catalogs results=\n\n{pformat(result)}\n")
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py:from xbr import sign_eip712_catalog_create
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: catalog_id = uuid.uuid4().bytes
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: signature = sign_eip712_catalog_create(
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: wallet_raw, verifyingChain, verifyingContract, wallet_adr, created, catalog_id, terms_hash, meta_hash
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: "xbr.network.create_catalog",
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: catalog_id,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: self.log.info("create_catalog results:\n\n{result}\n", result=pformat(result))
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: assert "action" in result and result["action"] == "create_catalog"
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: self.log.info('Create catalog - verification "{vaction_oid}" created', vaction_oid=vaction_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: fn = "create-catalog-email-verification.{}".format(vaction_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: result = await self.call("xbr.network.verify_create_catalog", vaction_oid.bytes, vaction_code)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: "catalog_oid" in result
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: and isinstance(result["catalog_oid"], bytes)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: and len(result["catalog_oid"]) == 16
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: catalog_oid = result["catalog_oid"]
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: "SUCCESS! New XBR FbsRepository Created: member_oid={member_oid}, catalog_oid={catalog_id}, "
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: catalog_id=uuid.UUID(bytes=catalog_oid).__str__(),
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: # Lets see if we can now fetch the newly created catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: result = await self.call("xbr.network.get_catalog", catalog_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api08_catalog.py: assert "oid" in result and isinstance(result["oid"], bytes) and result["oid"] == catalog_oid
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: # Get catalogs of the current user
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: catalogs_oids = await self.call("xbr.network.get_catalogs_by_owner", member_id.bytes)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: self.log.info(f"get_catalogs_by_owner results=\n\n{pformat(catalogs_oids)}\n")
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: assert len(catalogs_oids) > 0, "Not catalogs found for the current user"
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: catalog_id = catalogs_oids[0]
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: catalog_id,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: catalog_id,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: self.log.info("create_catalog results:\n\n{result}\n", result=pformat(result))
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: "catalog_oid" in result
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: and isinstance(result["catalog_oid"], bytes)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: and len(result["catalog_oid"]) == 16
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: and result["catalog_oid"] == catalog_id
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: catalog_oid = result["catalog_oid"]
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: "SUCCESS! New XBR Api published: api_oid={api_oid}, catalog_oid={catalog_id}, result: {result}\n",
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: catalog_id=uuid.UUID(bytes=catalog_oid).__str__(),
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: # Lets see if we can now fetch the newly created catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/network/test/test_api10_api.py: assert "catalog_oid" in result and isinstance(result["catalog_oid"], bytes)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: 'catalogs': 0,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: @wamp.register("xbr.network.create_catalog", check_types=True)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: async def create_catalog(
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: catalog_oid: bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param member_oid: OID of the member to create the catalog under (the member will become catalog owner.).
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param catalog_oid: OID of the new catalog.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param terms_hash: Multihash for optional catalog terms that apply to the catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: and all APIs published to that catalog.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param meta_hash: Mutlihash for optional off-chain catalog meta-data.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param meta_data: Optional off-chain catalog meta-data.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param signature: EIP712 signature for the catalog market creation.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :return: Data catalog creation information.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: assert is_bytes16(catalog_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: raise RuntimeError("Can only create catalog for own self!")
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: _catalog_oid = uuid.UUID(bytes=catalog_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: submitted = await self._network.create_catalog(
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: _catalog_oid,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: @wamp.register("xbr.network.verify_create_catalog", check_types=True)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: async def verify_create_catalog(
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: Verify creating a new data API catalog by submitting a verification code.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: Upon successful catalog creation, this procedure will also publish the same data that is returned
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: * **Procedure**: ``xbr.network.verify_create_catalog``
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: * **Events**: ``xbr.network.on_new_catalog``
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: 'catalog_oid': b'...',
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: * ``catalog_oid``: ID of newly created catalog (16 bytes UUID).
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: created_catalog_request_verified = await self._network.verify_create_catalog(vaction_oid, vaction_code)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: "xbr.network.on_catalog_created",
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: created_catalog_request_verified,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: return created_catalog_request_verified
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: @wamp.register("xbr.network.remove_catalog", check_types=True)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: def remove_catalog(
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :return: Data catalog removal information.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: @wamp.register("xbr.network.get_catalogs_by_owner", check_types=True)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: def get_catalogs_by_owner(self, member_oid: bytes, details: Optional[CallDetails] = None):
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param member_oid: OID of the member to retrieve owned catalogs for.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :return: List of OIDs of catalogs owned by the given member.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: raise RuntimeError("only own catalogs can be accessed!")
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: catalogs = []
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: for catalog_oid in self._xbr.idx_catalogs_by_owner.select(
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: catalogs.append(catalog_oid.bytes)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: return catalogs
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: @wamp.register("xbr.network.get_catalog", check_types=True)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: def get_catalog(
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: self, catalog_oid: bytes, include_attributes: bool = False, details: Optional[CallDetails] = None
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param catalog_oid: OID of the XBR Data FbsRepository to retrieve information for.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: _catalog_oid = uuid.UUID(bytes=catalog_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: raise ApplicationError("wamp.error.invalid_argument", "invalid _catalog_oid: {}".format(str(e)))
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: catalog = self._network.get_catalog(_catalog_oid, include_attributes)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: return catalog.marshal()
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: @wamp.register("xbr.network.find_catalogs", check_types=False)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: async def find_catalogs(
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: * APIs published to catalogs
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: as well as specify range and limit of the searched blockchain blocks and returned catalogs.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: When a specific filter is not provided, the filter remains un-applied and respective catalogs
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: will return *all* existing catalogs. The pagination via ``created_from`` and ``limit`` still applies.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param created_from: Only return catalogs created within blocks not earlier than this block number.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param limit: Only return catalogs from this many blocks beginning with block ``created_from``.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: To search for catalogs, the following filters can be used:
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param include_owners: If provided, only return catalogs owned by any of the owners specified.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param include_apis: If provided, only return catalogs containing any of the APIs specified.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param include_titles: If provided, only return catalogs with a title that
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param include_descriptions: If provided, only return catalogs with a description that
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param include_tags: If provided, only return catalogs with a tag that contains any of the specified tags.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :return: List of OIDs of catalogs matching the search criteria.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: catalog_oids = []
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: for catalog_oid in self._xbr.catalogs.select(txn, return_keys=True, return_values=False):
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: catalog_oids.append(catalog_oid.bytes)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: if len(catalog_oids) >= limit:
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: return catalog_oids
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: catalog_oid: bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param member_oid: OID of the member to create the catalog under (the member will become catalog owner.).
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param catalog_oid: OID of the (existing) catalog to publish the API to.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param api_oid: OID of the new API (as published in the catalog).
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param signature: EIP712 signature for the catalog market creation.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: assert isinstance(catalog_oid, bytes) and len(catalog_oid) == 16
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: catalog_oid_ = uuid.UUID(bytes=catalog_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: catalog_oid_,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: # catalog_oid = uuid.UUID(bytes=catalog_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: Verify publishing an API to an API catalog by submitting a verification code.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: Upon successful catalog creation, this procedure will also publish the same data that is returned
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: 'catalog_oid': b'...',
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: created_catalog_request_verified = self._network.verify_publish_api(vaction_oid, vaction_code)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: "xbr.network.on_api_published", created_catalog_request_verified, options=PublishOptions(acknowledge=True)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: return created_catalog_request_verified
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param api_oid: OID of the XBR API (as published in a catalog) to retrieve information for.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: include_catalogs: Optional[List[bytes]] = None,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: * catalog(s) the APIs are published in
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param include_catalogs: If provided, only return APIs published to a catalog in this list.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param include_titles: If provided, only return catalogs with a title that
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param include_descriptions: If provided, only return catalogs with a description that
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_api.py: :param include_tags: If provided, only return catalogs with a tag that contains any of the specified tags.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: recover_eip712_catalog_create,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: def _send_createCatalogFor(self, member, created, catalog_id, terms, meta, signature):
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: xbr.xbrcatalog.functions.createCatalogFor, member, created, catalog_id, terms, meta, signature
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: cnt_catalogs = 0
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: "catalogs": cnt_catalogs,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: async def create_catalog(
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog_oid: uuid.UUID,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: if not isinstance(catalog_oid, uuid.UUID):
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: raise RuntimeError("catalog_oid must be UUID, was {}".format(type(catalog_oid)))
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: signer_address = recover_eip712_catalog_create(
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog_oid.bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: "catalog_oid": catalog_oid.bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: self._save_verification_file(vaction.oid, "create-catalog-email-verification", vaction.verified_data)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: "action": "create_catalog",
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: async def verify_create_catalog(self, vaction_oid, vaction_code):
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: '{klass}.verify_create_catalog: invalid verification code "{vaction_code}" received in catalog creation - expected "{vaction_code_expected}"',
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog_oid = uuid.UUID(bytes=vaction.verified_data["catalog_oid"])
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog = Catalog()
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog.oid = catalog_oid
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog.timestamp = created
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog.owner = member_adr
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog.seq = vaction.verified_data["block_number"]
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: self._xbr.catalogs[txn, catalog.oid] = catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: self._xbr.idx_catalogs_by_owner[txn, (member_adr, created)] = member_oid
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: self._remove_verification_file(vaction_oid, "create-catalog-email-verification")
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog_oid = vaction.verified_data["catalog_oid"]
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: self._send_createCatalogFor, member, created, catalog_oid, terms_hash, meta, signature
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: "catalog_oid": catalog_oid,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: def get_catalog(self, catalog_oid, include_attributes):
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: assert isinstance(catalog_oid, uuid.UUID), 'catalog_oid must be bytes[16], was "{}"'.format(catalog_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog = self._xbr.catalogs[txn, catalog_oid]
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: if not catalog:
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: raise ApplicationError("crossbar.error.no_such_object", "no catalog {}".format(catalog_oid))
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: return catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog_oid,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: if not isinstance(catalog_oid, uuid.UUID):
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: raise RuntimeError("catalog_oid must be UUID, was {}".format(type(catalog_oid)))
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog_oid.bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: "catalog_oid": catalog_oid.bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: catalog_oid = uuid.UUID(bytes=vaction.verified_data["catalog_oid"])
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: api.catalog_oid = catalog_oid
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: self._xbr.idx_apis_by_catalog[txn, (catalog_oid.bytes, created)] = api_oid
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_backend.py: "catalog_oid": catalog_oid.bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_mailgw.py:_CREATE_CATALOG_VERIFICATION_EMAIL_TITLE = "XBR Network: confirm catalog creation {catalog_id}"
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_mailgw.py:XBR Network has received a request for you ({member_id}) to create an api catalog {catalog_id}.
/home/oberstet/work/wamp/crossbar/src/crossbar/network/_mailgw.py:_PUBLISH_API_VERIFICATION_EMAIL_TITLE = "XBR Network: confirm catalog creation {catalog_id}"
/home/oberstet/work/wamp/crossbar/src/crossbar/personality.py:from crossbar.webservice import archive, base, catalog, longpoll, misc, rest, static, wap, websocket, wsgi
/home/oberstet/work/wamp/crossbar/src/crossbar/personality.py: "catalogs": [
/home/oberstet/work/wamp/crossbar/src/crossbar/personality.py: "catalogs": [
/home/oberstet/work/wamp/crossbar/src/crossbar/personality.py: "catalog": catalog.RouterWebServiceCatalog.check,
/home/oberstet/work/wamp/crossbar/src/crossbar/personality.py: "catalog": catalog.RouterWebServiceCatalog,
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: catalog_oid = uuid.UUID(bytes=args.apiCatalog)
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: consent = self._xbr.consents[txn, (catalog_oid, member, delegate, delegate_type, market_oid)]
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: # Event emitted when a new API catalog is created
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: # emit CatalogCreated(catalogId, created, catalogSeq, owner, terms, meta);
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: catalog_oid = uuid.UUID(bytes=args.catalogId)
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: catalog = cfxdb.xbr.catalog.Catalog()
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: catalog.oid = catalog_oid
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: catalog.timestamp = created
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: catalog.seq = args.catalogSeq
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: catalog.owner = owner
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: catalog.terms = args.terms
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: catalog.meta = args.meta
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: self._xbr.catalogs[txn, catalog_oid] = catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: (xbr.xbrcatalog.events.CatalogCreated, _process_Catalog_CatalogCreated),
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: (xbr.xbrcatalog.events.ApiPublished, _process_Catalog_ApiPublished),
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbrmm.py: xbr.xbrcatalog.address,
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: "xbrcatalog": str(xbr.xbrcatalog.address),
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: catalog_oid: bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: XBR data catalog. The data is provided or consumed under the respective
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: data market terms and data catalog terms.
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: consumer and provider, and also for different API catalogs and markets.
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: :param catalog_oid: The OID of the XBR API catalog consent is given for. Depending on
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: API from the catalog.
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: services implementing the APIs in the catalog. This URI must be globally unique
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: _catalog_oid = uuid.UUID(bytes=catalog_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: _catalog_oid.bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: _catalog_oid.bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: consent_.catalog_oid = _catalog_oid
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: self._xbr.consents[txn, (_catalog_oid, member_adr, delegate, delegate_type, _market_oid)] = consent_
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: catalog_oid: bytes,
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: assert isinstance(catalog_oid, bytes) and len(catalog_oid) == 16
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: _catalog_oid = uuid.UUID(bytes=catalog_oid)
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: consent = self._xbr.consents[txn, (_catalog_oid, member, delegate, delegate_type, _market_oid)]
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: as well as specify range and limit of the searched blockchain blocks and returned catalogs.
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: async def get_catalogs_by_owner(self, owner_adr: bytes, details: Optional[CallDetails] = None):
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: :param owner_adr: Wallet address of the owner of the catalogs
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: :return: List of OIDs of catalogs matching the search criteria.
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: catalogs = self._xbr.idx_catalogs_by_owner.select(
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: for catalog in catalogs:
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: _catalog = self._xbr.catalogs[txn, catalog]
/home/oberstet/work/wamp/crossbar/src/crossbar/edge/worker/xbr/_marketmaker.py: marshaled = _catalog.marshal()
/home/oberstet/work/wamp/crossbar/src/crossbar/router/router.py: # setup optional inventory for realm API catalogs
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: # catalog metadata
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: raise RuntimeError('cannot open catalog from bfbs file "{}" - not a file'.format(filename))
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: catalog = Catalog(inventory=inventory, ctype=Catalog.CATALOG_TYPE_BFBS, name=name, bfbs=filename)
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: return catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: raise RuntimeError('cannot open catalog from archive "{}" - path is not a file'.format(filename))
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: raise RuntimeError('cannot open catalog from archive "{}" - file is not a ZIP file'.format(filename))
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: raise RuntimeError('cannot open catalog from archive "{}" - ZIP file is corrupt'.format(filename))
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: if "catalog.yaml" not in f.namelist():
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: raise RuntimeError("archive does not seem to be a catalog - missing catalog.yaml catalog index")
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: # open, read and parse catalog metadata file
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: data = f.open("catalog.yaml").read()
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: assert schema_path in f.namelist(), 'cannot find schema path "{}" in catalog archive'.format(
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: assert logo_path in f.namelist(), 'cannot find theme logo path "{}" in catalog archive'.format(
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: catalog = Catalog(
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: return catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: catalog = Catalog(
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: return catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: def __init__(self, personality, factory, catalogs: Optional[Dict[str, Catalog]] = None):
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: :param catalogs:
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: self._catalogs: Dict[str, Catalog] = catalogs or {}
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: # the consolidated schema repository with all schemas from catalogs
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: return len(self._catalogs)
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: return self._catalogs[name]
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: return iter(self._catalogs)
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: def add_catalog(self, catalog: Catalog):
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: :param catalog:
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: assert catalog.name not in self._catalogs
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: self._catalogs[catalog.name] = catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: if catalog.ctype == Catalog.CATALOG_TYPE_BFBS:
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: self._repo.load(catalog.bfbs)
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: def catalog(self, name: str):
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: return self._catalogs.get(name, None)
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: assert name not in self._catalogs
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: self._catalogs[name] = Catalog(inventory=self, ctype=Catalog.CATALOG_TYPE_BFBS, name=name, bfbs=filename)
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: assert "catalogs" in config and isinstance(config["catalogs"], list)
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: catalogs = {}
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: for idx, catalog_config in enumerate(config["catalogs"]):
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: if "bfbs" in catalog_config:
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: catalog = Catalog.from_bfbs(
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: name=catalog_config.get("name", "catalog{}".format(idx)),
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: filename=catalog_config["bfbs"],
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: elif "archive" in catalog_config:
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: catalog = Catalog.from_archive(inventory=inventory, filename=catalog_config["archive"])
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: elif "address" in catalog_config:
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: catalog = Catalog.from_address(inventory=inventory, address=catalog_config["address"])
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: assert False, 'neither "bfbs", "archive" nor "address" field in catalog config'
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: catalogs[catalog.name] = catalog
/home/oberstet/work/wamp/crossbar/src/crossbar/router/inventory.py: inventory._catalogs = catalogs
/home/oberstet/work/wamp/crossbar/src/crossbar/router/dealer.py: # 'catalog': 'pydefi.eth',
/home/oberstet/work/wamp/crossbar/src/crossbar/webservice/catalog.py: "/": "wamp_catalog_home.html",
/home/oberstet/work/wamp/crossbar/src/crossbar/webservice/catalog.py: "table": "wamp_catalog_table.html",
/home/oberstet/work/wamp/crossbar/src/crossbar/webservice/catalog.py: "struct": "wamp_catalog_struct.html",
/home/oberstet/work/wamp/crossbar/src/crossbar/webservice/catalog.py: "enum": "wamp_catalog_enum.html",
/home/oberstet/work/wamp/crossbar/src/crossbar/webservice/catalog.py: "service": "wamp_catalog_service.html",
/home/oberstet/work/wamp/crossbar/src/crossbar/webservice/catalog.py: if config["type"] != "catalog":
/home/oberstet/work/wamp/crossbar/src/crossbar/webservice/catalog.py: # must be equal to "catalog"
/home/oberstet/work/wamp/crossbar/src/crossbar/webservice/catalog.py: personality.WEB_SERVICE_CHECKERS["catalog"](personality, config)
oberstet@amd-ryzen5:~/scm/typedefint/wampire$
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment