In the previous section, we used the CopyToTable class as the template instead of luigi.Task. In fact, this is a good pattern to use! If there is any custom configuration or code you can use from one task to another, feel free to create a custom task class of your own. For example, in our practice, we use a custom S3Task class, similar to the one that follows:
from luigi.contrib.s3 import S3Client, S3Target
import pandas as pd
from io import StringIO, BytesIO
class S3Task(luigi.Task):
client = S3Client()
def _upload_csv(df, path):
content = df.to_csv(float_format="%.3f", index=None)
self.client.put_string(
content=content, destination_s3_path=path,
ContentType="text/csv"
)
def _upload_binary(self, df):
format_ = path.split(".")[-1]
...