src/astro/sql/operators/agnostic_load_file.py
def execute(self, context):
        """Loads csv/parquet table from local/S3/GCS with Pandas.

        Infers SQL database type based on connection then loads table to db.
        """

        if self.file_conn_id:
            BaseHook.get_connection(self.file_conn_id)

        # Retrieve conn type
        conn = BaseHook.get_connection(self.output_table.conn_id)
        if type(self.output_table) == TempTable:
            self.output_table = self.output_table.to_table(
                create_table_name(context=context), SCHEMA
            )
        else:
            self.output_table.schema = self.output_table.schema or SCHEMA
        if not self.output_table.table_name:
            self.output_table.table_name = create_table_name(context=context)

        if_exists = self.if_exists
        paths = self.get_paths(self.path, self.file_conn_id)
        for path in paths:
            # Read file with Pandas load method based on `file_type` (S3 or local).
            df = self._load_dataframe(path)

            move_dataframe_to_sql(
                output_table_name=self.output_table.table_name,
                conn_id=self.output_table.conn_id,
                database=self.output_table.database,
                warehouse=self.output_table.warehouse,
                schema=self.output_table.schema,
                df=df,
                conn_type=conn.conn_type,
                user=conn.login,
                chunksize=self.chunksize,
                if_exists=if_exists,
            )
            if_exists = "append"
        self.log.info(f"returning table {self.output_table}")
        return self.output_table
Loads csv/parquet table from local/S3/GCS with Pandas.

        Infers SQL database type based on connection then loads table to db.
        
Loads csv/parquet table from local/S3/GCS with Pandas.
Infers SQL database type based on connection then loads table to db.
Loads csv / parquet table from local / S3 / GCS with Pandas . Infers SQL database type based on connection then loads table to db .
def execute(self, context):
        if self.file_conn_id:
            BaseHook.get_connection(self.file_conn_id)
        conn = BaseHook.get_connection(self.output_table.conn_id)
        if type(self.output_table) == TempTable:
            self.output_table = self.output_table.to_table(
                create_table_name(context=context), SCHEMA
            )
        else:
            self.output_table.schema = self.output_table.schema or SCHEMA
        if not self.output_table.table_name:
            self.output_table.table_name = create_table_name(context=context)
        if_exists = self.if_exists
        paths = self.get_paths(self.path, self.file_conn_id)
        for path in paths:
            df = self._load_dataframe(path)
            move_dataframe_to_sql(
                output_table_name=self.output_table.table_name,
                conn_id=self.output_table.conn_id,
                database=self.output_table.database,
                warehouse=self.output_table.warehouse,
                schema=self.output_table.schema,
                df=df,
                conn_type=conn.conn_type,
                user=conn.login,
                chunksize=self.chunksize,
                if_exists=if_exists,
            )
            if_exists = "append"
        self.log.info(f"returning table {self.output_table}")
        return self.output_table
def execute ( self , context ) : if self . file_conn_id : BaseHook . get_connection ( self . file_conn_id ) conn = BaseHook . get_connection ( self . output_table . conn_id ) if type ( self . output_table ) == TempTable : self . output_table = self . output_table . to_table ( create_table_name ( context = context ) , SCHEMA ) else : self . output_table . schema = self . output_table . schema or SCHEMA if not self . output_table . table_name : self . output_table . table_name = create_table_name ( context = context ) if_exists = self . if_exists paths = self . get_paths ( self . path , self . file_conn_id ) for path in paths : df = self . _load_dataframe ( path ) move_dataframe_to_sql ( output_table_name = self . output_table . table_name , conn_id = self . output_table . conn_id , database = self . output_table . database , warehouse = self . output_table . warehouse , schema = self . output_table . schema , df = df , conn_type = conn . conn_type , user = conn . login , chunksize = self . chunksize , if_exists = if_exists , ) if_exists = "append" self . log . info ( f"returning table {self.output_table}" ) return self . output_table
Loads csv/parquet table from local/S3/GCS with Pandas.
Loads csv / parquet table from local / S3 / GCS with Pandas .
"""Loads csv/parquet table from local/S3/GCS with Pandas.

        Infers SQL database type based on connection then loads table to db.
        """ # Retrieve conn type # Read file with Pandas load method based on `file_type` (S3 or local).
self
context