(gcs_client, set_name: str)
| 194 | |
| 195 | |
| 196 | def _get_usage_set(gcs_client, set_name: str) -> Set[str]: |
| 197 | try: |
| 198 | result = set() |
| 199 | usages = gcs_client.internal_kv_keys( |
| 200 | set_name.encode(), |
| 201 | namespace=usage_constant.USAGE_STATS_NAMESPACE.encode(), |
| 202 | ) |
| 203 | for usage in usages: |
| 204 | usage = usage.decode("utf-8") |
| 205 | result.add(usage[len(set_name) :]) |
| 206 | |
| 207 | return result |
| 208 | except Exception as e: |
| 209 | logger.debug(f"Failed to get usage set {set_name}, {e}") |
| 210 | return set() |
| 211 | |
| 212 | |
| 213 | def _put_library_usage(library_usage: str): |
no test coverage detected
searching dependent graphs…