Reloading a code location with code or requests

[axellpadilla]

Here is an example on how to reload current location, this helps doing some things that aren't posible otherwise (if this could be simplified let me know):

• Allows to update external definitions (for example using sling or DLT) using an op or asset.
• Regenerate .yml files or structures that are later used in an asset factory and make them load.
• Reload location instantly after simply code changes trough GitHub actions.
• Make initial full reload faster using conditional caches when relying on external sources schemas.

Use with caution, centralize and test it because is an internal API and could require changes when upgrading, I hope dagster team address this and includes as a formal code API to be able to use it on code.

def reload_code_location(host: str, port: int, location_name: str) -> dict:
    """
    Sends a GraphQL mutation to the Dagster webserver to reload the specified repository location.

    :param host: The hostname or IP of the Dagster webserver (e.g. 'localhost', '100.#.#.#').
    :param port: The port the webserver is running on (e.g. 9786).
    :param location_name: The code location name to reload.
    :return: The JSON response from the Dagster webserver.
    """

    # This is the GraphQL mutation string used to reload a repository location.
    graphql_mutation = """
    mutation ReloadRepositoryLocationMutation($location: String!) {
      reloadRepositoryLocation(repositoryLocationName: $location) {
        ... on WorkspaceLocationEntry {
          id
          loadStatus
          locationOrLoadError {
            ... on RepositoryLocation {
              id
              __typename
            }
            ... on PythonError {
              message
              stack
            }
            __typename
          }
          __typename
        }
        ... on UnauthorizedError {
          message
          __typename
        }
        ... on ReloadNotSupported {
          message
          __typename
        }
        ... on RepositoryLocationNotFound {
          message
          __typename
        }
        ... on PythonError {
          message
          stack
        }
        __typename
      }
    }
    """

    variables = {"location": location_name}

    # GraphQL endpoint; note ?op=... is optional but matches what the UI does
    url = f"http://{host}:{port}/graphql?op=ReloadRepositoryLocationMutation"

    # For local/unsecured development, these headers are often enough. Adjust as needed for auth.
    headers = {
        "content-type": "application/json",
        "accept": "*/*",
    }

    payload = {
        "operationName": "ReloadRepositoryLocationMutation",
        "variables": variables,
        "query": graphql_mutation,
    }

    response = requests.post(url, headers=headers, json=payload, verify=False)
    response.raise_for_status()  # Raise exception if the request failed at the HTTP level

    return response.json()