Custom XCom GCS

[rajaths010494]

Custom XCOM to upload xcom values to the GCS bucket for the given name.
Adding ENV AIRFLOW__CORE__XCOM_BACKEND: astronomer.providers.google.cloud.xcom_backends.gcs.GCSXComBackend in the docker-compose.yaml or where the airflow is running will enable the custom XCOM for GCS.

closes #596

[codecov]

Codecov Report

Base: 98.47% // Head: 98.49% // Increases project coverage by +0.02% 🎉

Coverage data is based on head (06c4d87) compared to base (99dd72d).
Patch coverage: 100.00% of modified lines in pull request are covered.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #681      +/-   ##
==========================================
+ Coverage   98.47%   98.49%   +0.02%     
==========================================
  Files          86       87       +1     
  Lines        4661     4724      +63     
==========================================
+ Hits         4590     4653      +63     
  Misses         71       71              

Impacted Files	Coverage Δ
...onomer/providers/google/cloud/xcom_backends/gcs.py	100.00% <100.00%> (ø)

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

☔ View full report at Codecov.
📢 Do you have feedback about the report comment? Let us know in this issue.

[uranusjr]

I would restructure the implementation like this:

# Only implements the required interface and "forward" logic.
class GCSXCom(BaseXCom):
    def orm_deserialize_value(self) -> str:
        ...

    @staticmethod
    def serialize_value(value: Any, **kwargs) -> Any:
        value = _GCSXComBackend().write_and_upload_value(value, **kwargs)
        return BaseXCom.serialize_value(value)

    @staticmethod
    def deserialize_value(xcom: "XCom") -> Any:
        result = BaseXCom.deserialize_value(xcom)
        if isinstance(result, str) and result.startswith(GCSXComBackend.PREFIX):
            result = _GCSXComBackend().download_and_read_value(result)
        return result

# Implements the actual logic!
class _GCSXComBackend:
    def write_and_upload_value(self, value, **kwargs):
        ...

    def download_and_read_value(self, result):
        ...

The advantage to this is we can reference values with self, which enables us to cache things (for example the GCSHook) and make the environment variable retrievals lazy properties. This reduces import-time logic (makes startup fast) and can also simplify testing.

[rajaths010494]

I would restructure the implementation like this:

# Only implements the required interface and "forward" logic.
class GCSXCom(BaseXCom):
    def orm_deserialize_value(self) -> str:
        ...

    @staticmethod
    def serialize_value(value: Any, **kwargs) -> Any:
        value = _GCSXComBackend().write_and_upload_value(value, **kwargs)
        return BaseXCom.serialize_value(value)

    @staticmethod
    def deserialize_value(xcom: "XCom") -> Any:
        result = BaseXCom.deserialize_value(xcom)
        if isinstance(result, str) and result.startswith(GCSXComBackend.PREFIX):
            result = _GCSXComBackend().download_and_read_value(result)
        return result

# Implements the actual logic!
class _GCSXComBackend:
    def write_and_upload_value(self, value, **kwargs):
        ...

    def download_and_read_value(self, result):
        ...

The advantage to this is we can reference values with self, which enables us to cache things (for example the GCSHook) and make the environment variable retrievals lazy properties. This reduces import-time logic (makes startup fast) and can also simplify testing.

Yes, i am doing it. Will need to modify test as well for the changes

[bharanidharan14]

@kaxil @uranusjr @josh-fell Need your review on this PR

[kaxil]

Unrelated changes?

[rajaths010494]

Rebased branch