def _init_default_session_from_env()

in pai/session.py [0:0]

• 57 lines of code
• 9 McCabe index (conditional complexity)

def _init_default_session_from_env() -> Optional["Session"]:
    credential_client = Session._get_default_credential_client()
    if not credential_client:
        logger.debug("Not found credential from default credential provider chain.")
        return

    # legacy region id env var in DSW
    region_id = os.getenv("dsw_region")
    region_id = os.getenv("REGION", region_id)
    if not region_id:
        logger.debug(
            "No region id found(env var: REGION or dsw_region), skip init default session"
        )
        return

    dsw_instance_id = os.getenv("DSW_INSTANCE_ID")
    if not dsw_instance_id:
        logger.debug(
            "No dsw instance id (env var: DSW_INSTANCE_ID) found, skip init default session"
        )
        return

    workspace_id = os.getenv("PAI_AI_WORKSPACE_ID")
    workspace_id = os.getenv("PAI_WORKSPACE_ID", workspace_id)

    network = (
        Network.VPC
        if is_domain_connectable(
            PAI_VPC_ENDPOINT.format(region_id),
            timeout=1,
        )
        else Network.PUBLIC
    )

    if dsw_instance_id and not workspace_id:
        logger.debug("Getting workspace id by dsw instance id: %s", dsw_instance_id)
        workspace_id = Session._get_workspace_id_by_dsw_instance_id(
            dsw_instance_id=dsw_instance_id,
            cred=credential_client,
            region_id=region_id,
            network=network,
        )
        if not workspace_id:
            logger.warning(
                "Failed to get workspace id by dsw instance id: %s", dsw_instance_id
            )
            return
    bucket_name, oss_endpoint = Session.get_default_oss_storage(
        workspace_id, credential_client, region_id, network
    )

    if not bucket_name:
        logger.warning(
            "Default OSS storage is not configured for the workspace: %s", workspace_id
        )

    sess = Session(
        region_id=region_id,
        workspace_id=workspace_id,
        credential_config=None,
        oss_bucket_name=bucket_name,
        oss_endpoint=oss_endpoint,
        network=network,
    )

    return sess