Google Imports are Tied Together

[dylanbhughes]

Description

A clear description of the bug
I’m using the new GCSResult and I’m getting an import error when I don’t also specify google-cloud-bigquery as a dependency since they’re imports occur in the same file, I think?

Unexpected error: ImportError("cannot import name 'bigquery' from 'google.cloud' (unknown location)")
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 986, in get_task_run_state
    result = self.result.write(value, filename="output", **prefect.context)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/results/gcs_result.py", line 73, in write
    self.gcs_bucket.blob(new.location).upload_from_string(binary_data)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/results/gcs_result.py", line 35, in gcs_bucket
    from prefect.utilities.gcp import get_storage_client
  File "/usr/local/lib/python3.7/site-packages/prefect/utilities/gcp.py", line 6, in <module>
    from google.cloud import bigquery, storage
ImportError: cannot import name 'bigquery' from 'google.cloud' (unknown location)

https://cloud.prefect.io/prefect/flow-run/6704aa4e-ba9b-40ed-a4f8-386920839a8e?logId=75b1fc01-0ee8-4061-ab8b-5481e6123a79

On a cool note, changing to python_dependencies=["prefect[google]"] did work 🎉

Expected Behavior

What did you expect to happen instead?
I'd like to be able to specify one import in insolation (in this case google-cloud-storage)

Reproduction

A minimal example that exhibits the behavior.

from prefect import task, Flow
from prefect.tasks.notifications.slack_task import SlackTask
from prefect.schedules import CronSchedule
from prefect.environments.storage import Docker
from prefect.engine.results import GCSResult

import pendulum
import datetime


@task(name="Get Week Message", max_retries=5, retry_delay=datetime.timedelta(seconds=5))
def get_week_message():
    prefects_birthday = pendulum.date(2018, 1, 17)
    current_week = prefects_birthday.diff(pendulum.now()).in_weeks()
    return f"Hello, Jeremiah! It is week {current_week}."


send_message = SlackTask(
    name="Slack Jeremiah",
    max_retries=5,
    retry_delay=datetime.timedelta(seconds=5),
    webhook_secret="SLACK_WEBHOOK",
)

schedule = CronSchedule(cron="50 11 * * MON", start_date=pendulum.now(tz="US/Eastern"))

storage = Docker(
    base_image="prefecthq/prefect:latest-python3.7",
    registry_url=URL,
    python_dependencies=["google-cloud-storage"],
    files={
        FILE_LOCATION: FILENAME
    },
    env_vars={"GOOGLE_APPLICATION_CREDENTIALS": FILENAME},
)

gcs_result = GCSResult(bucket="what_week_is_it_results")

with Flow(
    name="What Week is It?", schedule=schedule, storage=storage, result=gcs_result
) as flow:
    week_message = get_week_message()
    result = send_message(message=week_message)

Environment

Any additional information about your environment

Optionally run prefect diagnostics from the command line and paste the information here

{
  "config_overrides": {
    "cloud": {
      "auth_token": true,
      "use_local_secrets": true
    },
    "context": {
      "secrets": false
    },
    "home_dir": true
  },
  "env_vars": [],
  "system_information": {
    "platform": "Darwin-19.4.0-x86_64-i386-64bit",
    "prefect_version": "0.11.2",
    "python_version": "3.7.7"
  }
}