docstr: Patches the API to log media or metrics to W&B.
docstr_tok: ["Patches","the","API","to","log","media","or","metrics","to","W","&","B","."]
code: def patch(self, run: "wandb.sdk.wandb_run.Run") -> None:
      """Patches the API to log media or metrics to W&B."""
      for symbol in self.symbols:
          # split on dots, e.g. "Client.generate" -> ["Client", "generate"]
          symbol_parts = symbol.split(".")
          # and get the attribute from the module
          original = functools.reduce(
              getattr, symbol_parts, self.set_api
          )
  
          def method_factory(original_method: Any):
              async def async_method(*args, **kwargs):
                  future = asyncio.Future()
  
                  async def callback(coro):
                      try:
                          result = await coro
                          loggable_dict = self.resolver(
                              args,
                              kwargs,
                              result,
                              timer.start_time,
                              timer.elapsed,
                          )
                          if loggable_dict is not None:
                              run.log(loggable_dict)
                          future.set_result(result)
                      except Exception as e:
                          logger.warning(e)
  
                  with Timer() as timer:
                      coro = original_method(*args, **kwargs)
                      asyncio.ensure_future(callback(coro))
  
                  return await future
  
              def sync_method(*args, **kwargs):
                  with Timer() as timer:
                      result = original_method(*args, **kwargs)
                      try:
                          loggable_dict = self.resolver(
                              args,
                              kwargs,
                              result,
                              timer.start_time,
                              timer.elapsed,
                          )
                          if loggable_dict is not None:
                              run.log(loggable_dict)
                      except Exception as e:
                          logger.warning(e)
                      return result
  
              if inspect.iscoroutinefunction(original_method):
                  return functools.wraps(original_method)(async_method)
              else:
                  return functools.wraps(original_method)(sync_method)
  
          # save original method
          self.original_methods[symbol] = original
          # monkey patch the method
          if len(symbol_parts) == 1:
              setattr(
                  self.set_api,
                  symbol_parts[0],
                  method_factory(original),
              )
          else:
              setattr(
                  functools.reduce(
                      getattr, symbol_parts[:-1], self.set_api
                  ),
                  symbol_parts[-1],
                  method_factory(original),
              )
  
imports: import asyncio
  import functools
  import inspect
  import logging
  import sys
  from typing import Any, Dict, Optional, Sequence, TypeVar
  import wandb.sdk
  import wandb.util
  from wandb.sdk.lib import telemetry
  from wandb.sdk.lib.timer import Timer
label_desc: Use image node_id 3 for calling the PatchAPI obj's underlying member method code with example usage: obj.patch(run) and returns: unknown, functools, functools, result