Automate file processing with Buckets and Objects Part II

Automate receiving files from various sources, processing them, and sending them to external destinations.


18 min read

(Part II of a series on Putting Buckets and Objects to work. Part I Putting Oracle Buckets and Objects To Work created the foundation for viewing and managing Buckets in an APEX application unifying different data sources and the user-interface in a scalable, versatile manner.)


Depending on the quantity and complexity of data processed from and to external sources achieving productivity, consistency, accuracy, and security can be challenging. Secure FTP (sFTP) is one channel that acts as a gateway for sending, receiving, and staging/storing files through Buckets and Objects. REST Data Sources and websites are other data channel examples.

The source code presented below is not production ready but is intended for illustrative purposes. It does not include error handling, logging, or security authorization, and in some cases are code snippets, recognizing different standards among organizations and for brevity.


Automate the receiving, sending, and processing of files utilizing OCI Buckets, Objects and APEX in a secure, consistent manner that increases productivity using sFTP as an example source and destination.


  • Productivity: Increase productivity by automating data transmission and processing thereby reducing required effort.
  • Consistency and Accuracy: Increase consistency and accuracy with automated processes and workflow following the same rules each time.
  • Data Driven: Minimize "hard coding" with data driven processing based on Object Registry attributes. Changes to Object attributes such as database instance, process to run, whether it should be processed (an on/off switch), credentials, full URIs, etc. can be queried by processes in the database and are updateable through an APEX application instead of needing changes to program source code.
  • Flexibility: Adding new files or adjusting to modifications of Originator or Destination requirements are more likely handled in the user-interface, for example adding an entry for another Marketing file to import from the Market Analytics firm.
  • Insight: Access to the data through an APEX user-interface makes it easier to determine what processes are in place, log history, and overall relationships.


Let's begin with the architecture described in Part I of the series, "Putting Oracle Buckets and Objects To Work".


In brief, the table XE_Buckets unifies Buckets from two different regions using REST data sources and synchronization with tables that mirror REST data source structures and values. User-defined columns exist in XE_Buckets, not the synchronization tables. This Blog assumes that architecture is in place.

The architecture below picks up where Part I left off integrating sFTP, OCI Buckets, Objects, processes, and an APEX application.

600_FTP Architecture.jpg

The "Data Originator" transmits files via sFTP to the sFTP Server. Command Line Interface (CLI) Oracle CLI scripts regularly running on the sFTP Server forward the files to pre-designated OCI Buckets depending on file type and removes them from the inbound sftp folders to avoid duplication. Data sources and destinations could have been a(n) REST Service(s), website, etc. Here sFTP will be used as the example.

Inbound development, test, and production buckets named ftp_inbound_dev, ftp_inound_test, and ftp_inbound, respectively, exist to receive the files. Each instance can have their own copy to facilitate usage without impacting the other. Below is a sample excerpt from the XE_Buckets view.


The Object Registry describes inbound and outbound files, user-defined attributes, and processing information. To be clear, there is one descriptive entry for each file type, not an entry for each actual file. The Object Log tracks the actual files themselves. The Registry makes processing as data driven as possible, enables process updating through an APEX interface, and provides a level of documentation.


The Object Registry table structure is:

    XE_BUCKETS_ID                 NUMBER(20)          Constraint XE_OCI_OBJECT_REGISTRY_BU_NN NOT NULL,
    OP_XE_BUCKETS_ID_MOVE_TO      NUMBER(20)          ,
    OP_FILE_EXTENSION             VARCHAR2(50 CHAR)   ,
    OP_FILE_DIRECTION_CID         NUMBER(20)     ,
    OP_DATA_ROW_ACTION_CID        NUMBER(20)     , -- i.e. Delete, Import, Export, Noaction
    OP_SY_ENTITY_ID_SOURCE       NUMBER(20)     ,
    OP_SY_ENTITY_ID_DEST         NUMBER(20)     ,
    OP_OBJECT_PROCESS_CID         NUMBER(20)          ,
    OP_PROCESS_YN                 VARCHAR2(1 CHAR)    ,
    OP_OVERWRITE_YN              VARCHAR2(1 CHAR),
    OP_EFFECTIVE                  DATE        Default on Null Trunc(Current_Date) CONSTRAINT XE_OCI_OBJECT_REGISTRY_EE_NN  NOT NULL ,
    OP_EXPIRES                    DATE        Default on Null To_Date('12/31/2999','MM/DD/YYYY') CONSTRAINT XE_OCI_OBJECT_REGISTRY_EX_NN    NOT NULL ,
      SY_USER_ID_MOD                NUMBER(20)  ,
       SY_DATE_MOD                   DATE        ,
      Constraint                    XE_OCI_OBJECT_REGISTRY_BU_FK Foreign Key ( XE_BUCKETS_ID )          References XE_BUCKETS(XE_BUCKETS_ID),
      Constraint                    XE_OCI_OBJECT_REGISTRY_CD_FK Foreign Key ( 
   OP_CURRENT_DATABASE_CID) References Codes_Value(Codes_Value_ID),
      Constraint                    XE_OCI_OBJECT_REGISTRY_DT_FK Foreign Key (     OP_FILE_DATA_TYPE_CID )  References CODES_VALUE(CODES_VALUE_ID),
      Constraint                    XE_OCI_OBJECT_REGISTRY_FD_FK Foreign Key ( OP_FILE_DIRECTION_CID )  References CODES_VALUE(CODES_VALUE_ID),
      Constraint                    XE_OCI_OBJECT_REGISTRY_RA_FK Foreign Key ( OP_DATA_ROW_ACTION_CID ) References CODES_VALUE(CODES_VALUE_ID),
      Constraint                    XE_OCI_OBJECT_Registry_OP_FK Foreign Key ( OP_OBJECT_PROCESS_CID )  References Codes_Value(CODES_VALUE_ID),
      Constraint                    XE_OCI_OBJECT_REGISTRY_VS_FK Foreign Key ( OP_SY_ENTITY_ID_SOURCE ) References SY_ENTITY(SY_ENTITY_ID),
      Constraint                    XE_OCI_OBJECT_REGISTRY_VD_FK Foreign Key ( OP_SY_ENTITY_ID_DEST )  References SY_ENTITY(SY_ENTITY_ID),
      Constraint                    XE_OCI_OBJECT_REGISTRY_OP_PK Primary Key (XE_OCI_OBJECT_REGISTRY_ID)

Notes on the Registry columns are:

  • Processing needs vary across organizations. The objective here is to list possible user-defined columns that may be generally appropriate, and to inspire thought about what other attributes might be helpful in making the processing data driven and more easily maintained in your organization. Alternatively, these or other attributes could be tracked in JSON formatted column.
  • Columns ending in CID: "CID" columns contain primary keys of code values found in a Code Value view as described in Simplyfy managing codes and their values In short, instead of having many tables for each code and value combination, Code Type, Code Description, and Code Value tables provide a vehicle for managing them in three tables and a view and an APEX application user-interface only. Or, use your own methods for tracking codes and their values.
  • XE_Buckets_ID: Unique Bucket ID the object initially appears in.
  • OP_Object_Name: Case sensitive Object name. Use APEX_Util.URL_Encode() when referring to the item directly in the bucket to address the vagaries of file names.
  • OP_XE_Buckets_ID_Move_To: ID of Bucket where the Object should be moved to once processing is complete, removing it from Bucket XE_Buckets_ID to avoid duplicate processing. If Object names repeat, consider enabling Bucket Versioning to keep history.
  • OP_Current_Database_CID: Database instance this entry pertains to, i.e. Dev, Test, Prod. While creation of an entry for a single file for each instance leads to some duplication, it does allow for customization by instance. For example, files can have entries in Dev and Test before they are ready for production, or exist only in Dev and Test because they contain dummy test data that would not be appropriate for Production, etc. Alternatively, use a code value of "ANY" to indicate a match to all instances and only need one descriptive file entry. The Instance can be used to make refreshing Dev and Test with a data dump import from Prod easier. Attributes such as starting and move to bucket IDs do not need to be changed to Dev and Test Buckets if there is an entry for each Instance, or no entry for a particular file's instance such as Prod if the file is still in development and test.
  • OP_File_Extension: File extension of Object
  • OP_File_Data_Type_CID: Describes contents of the Object, i.e. Payroll, Inventory, Sales, etc.
  • OP_File_Direction_CID: Inbound, Outbound, Stationary, etc.
  • OP_Data_Row_Action_CID: Action to perform on the Object, i.e. Import, Export, Process, etc.
  • OP_SY_Entity_ID_Source: The Entity originating the data.
  • OP_SY_Entity_ID_Dest: The Entity receiving the data
  • OP_Table_Imported_To: Internal table the data will be applied/added to
  • OP_Table_Exported_From: Internal table or view originating the data
  • OP_Object_Process_CID: Name or Abbreviation of Object Process used by processing code to determine which set of actions to take on the data.
  • OP_Process_YN: Switch to turn automated processing on/off
  • OP_OverWrite_YN: Overwrite existing objects or content if it already exists Yes/No.
  • OP_Effective and OP_Expires: Date range in which this entry is valid.
  • SY Add and Mod: Track user and dates when this entry originally created and last updated.

Manage the Object Registry entries with a Modal Dialog Form linked with the Object Registry ID.


The code column naming convention is OP_ [Code Abbreviation] _CID. The resulting Source SQL Query for the Database Instance (Current_Database) item as an example is:

  Select CV_Description, Codes_Value_ID
        Current_Date Between CV_Effective and CV_Expires And
        Current_Date Between CO_Effective and CO_Expires
  Order By CV_Description

With a Default Value of:

  Select CV_Description, Codes_Value_ID
        Current_Date Between CV_Effective and CV_Expires And
        Current_Date Between CO_Effective and CO_Expires
  Order By CV_Description
  Fetch first 1 rows only;

The Table Imported To Source is a List of Values with a SQL Query of:

Select Table_Name Disp_Value, Table_Name
  From User_Tables
  Order by Table_Name

The Table Exported From Source is a List of Values with a SQL Query of:

Select Table_Name Disp_Value, Table_Name
  From User_Tables
  Select View_Name Disp_Value, View_Name Table_Name
    From User_Views
Order By Table_Name;

XE_Buckets, XE_OCI_Object_Registry, and Codes_Value are tied together in the following View.

Create Or Replace View VW_XE_OCI_Object_Registry
    ,(VW_XE_Buckets.BU_FULL_URI || XE_OCI_Object_Registry.OP_Object_Name) OP_FULL_Object_URI
    ,CD_Current_Database.CV_Value_Character DC_Current_Database
    ,XE_Buckets_Move.BU_Bucket_Name  BU_Bucket_Name_Move_To
    ,XE_Buckets_Move.BU_Credential  BU_Credential_Move_To
    ,XE_Buckets_Move.BU_FULL_URI    BU_Full_URI_Move_To
    ,(XE_Buckets_Move.BU_Full_URI || XE_OCI_Object_Registry.OP_Object_Name) OP_Full_Object_URI_Move_To
    ,CD_File_Data_Type.CV_Value_Character DC_File_Data_Type
    ,CD_File_Direction.CV_Value_Character DC_File_Direction
    ,CD_Data_Row_Action.CV_Value_Character DC_Data_Row_Action
    ,Entity_Source.EN_Abbrev EN_Abbrev_Source
    ,Entity_Dest.EN_Abbrev EN_Abbrev_Dest
    ,CD_Object_Process.CV_Value_Character DC_Object_Process

  From  XE_OCI_Object_Registry
       ,VW_XE_Buckets  XE_Buckets_Move
       ,Codes_Value       CD_Current_Database
       ,Codes_Value       CD_File_Data_Type
       ,Codes_Value       CD_File_Direction
       ,Codes_Value       CD_Data_Row_Action
       ,Codes_Value       CD_Object_Process
       ,SY_Entity            SY_Entity_Source
       ,SY_Entity            SY_Entity_Dest

  Where     XE_OCI_Object_Registry.XE_Buckets_ID            = VW_XE_Buckets.XE_Buckets_ID
        And Current_Date Between Trunc(VW_XE_Buckets.BU_Effective) and Trunc(VW_XE_Buckets.BU_Expires)
        And XE_OCI_Object_Registry.OP_XE_BUCKETS_ID_MOVE_TO = XE_Buckets_Move.XE_Buckets_ID (+)
        And XE_OCI_Object_Registry.OP_Current_Database_CID  = CD_Current_Database.Codes_Value_ID
        And XE_OCI_Object_Registry.OP_File_Data_Type_CID    = CD_File_Data_Type.Codes_Value_ID
        And XE_OCI_Object_Registry.OP_File_Direction_CID    = CD_File_Direction.Codes_Value_ID
        And XE_OCI_Object_Registry.OP_Data_Row_Action_CID   = CD_Data_Row_Action.Codes_Value_ID
        And XE_OCI_Object_Registry.OP_Object_Process_CID    = CD_Object_Process.Codes_Value_ID (+)
        And XE_OCI_Object_Registry.OP_SY_Entity_ID_Source  = Entity_Source.SY_Entity_ID
        And XE_OCI_Object_Registry.OP_SY_Entity_ID_Dest    = Entity_Dest.SY_Entity_ID
        And Current_Date Between OP_Effective and OP_Expires

Note, in addition to translating linked attributes between tables, the view includes full Bucket and Object URI's and appropriate credential names.

Next, create an Object Processing Log to track results as shown below.

The Object Processing Log captures actions taken, when, where the Object ended up, and other tracking information.


Table Creation:

    LO_OBJECT_NAME            Varchar2(256 CHAR) CONSTRAINT XE_OCI_Objects_Log_ON_NN Not Null,
    LO_CHECKSUM               Varchar2(256 CHAR) CONSTRAINT XE_OCI_Objects_Log_CK_NN Not Null,
        LO_XE_BUCKETS_ID_CURRENT  Number(20)        ,
        LO_OBJECT_CREATED         DATE              ,
        LO_BATCH_ID               NUMBER(20)        ,
        LO_FILE_EXTENSION         VARCHAR2(50 CHAR) ,
        LO_FILE_SIZE              Number(20)        ,
        LO_EFFECTIVE              DATE              Default on Null Trunc(Current_Date) CONSTRAINT XE_OCI_OBJECTS_LOG_EE_NN  NOT NULL ,
        LO_EXPIRES                DATE              Default on Null To_Date('12/31/2999','MM/DD/YYYY') CONSTRAINT XE_OCI_OBJECTS_LOG_EX_NN    NOT NULL ,
      SY_USER_ID_MOD            NUMBER(20)        ,
      SY_DATE_MOD               DATE              ,
      VC_LO_OBJECT_NAME_CAPS    Generated always as (Upper(LO_OBJECT_NAME)),
      VC_LO_FILE_EXTNESION_CAPS Generated always as (Upper(LO_FILE_EXTENSION)),
      Constraint                XE_OCI_OBJECTS_LOG_RA_FK Foreign Key ( LO_DATA_ROW_ACTION_CID ) References CODES_VALUE(CODES_VALUE_ID),
      Constraint                XE_OCI_OBJECTS_LOG_RS_FK Foreign Key ( LO_DATA_ROW_STATUS_CID ) References CODES_VALUE(CODES_VALUE_ID),
      Constraint                XE_OCI_OBJECTS_LOG_BF_FK Foreign Key ( LO_XE_BUCKETS_ID_CURRENT ) References XE_Buckets(XE_BUCKETS_ID),
      Constraint                XE_OCI_OBJECTS_LOG_BA_FK Foreign Key ( LO_BATCH_ID ) References SYBATCH(BATCH_ID),
      Constraint                XE_OCI_OBJECTS_LOG_LO_PK   Primary Key (XE_OCI_OBJECTS_LOG_ID)

Object Log View:

Create Or Replace View VW_XE_OCI_OBJECTS_LOG
    ,XE_BUCKETS.BU_Bucket_Name                  BU_Bucket_Name_Current
    ,XE_Buckets.BU_Bucket_Type                  BU_Bucket_Type_Current
    ,VW_XE_OCI_Object_Registry.BU_Bucket_Name   BU_Bucket_Name_Original
    ,VW_XE_OCI_Object_Registry.BU_Bucket_Type   BU_Bucket_Type_Original

  From  XE_OCI_Objects_Log
       ,Codes_Value DC_DATA_ROW_ACTION
       ,Codes_Value DC_DATA_ROW_STATUS
       ,VW_XE_Buckets XE_Buckets

            XE_OCI_Objects_Log.LO_XE_Buckets_ID_Current = XE_Buckets.XE_Buckets_ID
        And XE_OCI_Objects_Log.XE_OCI_Object_Registry_ID= VW_XE_OCI_Object_Registry.XE_OCI_Object_Registry_ID
        And XE_OCI_Objects_Log.LO_Data_Row_Action_CID   = DC_Data_Row_Action.Codes_Value_ID
        And XE_OCI_Objects_Log.LO_Data_Row_Status_CID   = DC_Data_Row_Status.Codes_Value_ID

Now for Import/Export processing. The following is based on packages, procedures, and functions being executed to carry out review of the Object Registry and processing on an ad hoc and APEX Application Shared Components Automations basis.

The possibilities seem endless... They can be scheduled through APEX Application Shared Component Automations, executed on an ad hoc basis, or triggered by Object Events Oracle Object Events. Functions could also be serverless: Oracle Serverless Functions An Oracle Workflow could be incorporated as well. But, it is unlikely you would be interested in reading about all of that in one Blog :)

We will stay with the ad-hoc and Shared Components Automations model for purposes of illustration.

  -- For each XE_OCI_Object_Registry for current DB instance, and processes
  --  Is object in XE_Buckets_ID Bucket, Process_YN = Y, Row effective, Same
  --     DB_Instance as Current?
  --    - Yes
  --      Run identified Object Process to
  --        Type is 'IMPORT'
  --          Import into designated OP_Table_Imported_To
  --          Copy Object to OP_XE_Buckets_ID_Move_TO Bucket
  --           Delete from XE_Buckets_ID Bucket
  --        Type is 'EXPORT'
  --           Export to designated Bucket
  --        Insert Entry into XE_OCI_Object_Log
  --    - No
  --      Null;
  Procedure Process_Objects( p_DC_File_Direction IN VW_XE_OCI_Object_Registry.DC_File_Direction%Type )
  v_Current_Database_CID      Number(20);
    v_Cloud_Object_Info       Pkg_XE_OCI_Object_Registry.t_Cloud_Object;
  v_Batch_ID                      Batch.Batch_ID%Type;
    v_DC_File_Direction         VW_XE_OCI_Object_Registry.DC_File_Direction%Type;
    v_OP_File_Direction_CID     Codes_Value.Codes_Value_ID%Type;
  v_XE_OCI_Objects_Log  XE_OCI_Objects_Log%RowType;
  v_Message                     App_Error.Ae_Message%Type;
  e_Error                        Exception;

      v_DC_File_Direction         := Upper(p_DC_File_Direction);
    -- Validate direction
    v_OP_File_Direction_CID     := Pkg_Codes_Value.Qry_Code_Value_To_ID(p_CO_NAME_ABREV                    => 'FILE_DIRECTION',
                                                                            p_CV_Value_Character        => v_DC_File_Direction,
                                                                            p_Error_on_No_Find        => 'Y');

-- Determine current database instance Codes_Value_ID
    v_Current_Database_CID:= Pkg_Codes_Value.Qry_Code_Value_To_ID(p_CO_Name_Abrev=> 'CURRENT_DATABASE',                                                           
                                           p_CV_Value_Character    =>    Current_DB_Instance);
    For c_Object_Registry In (Select *
                    From VW_XE_OCI_Object_Registry
                    Where     Current_Date Between VW_XE_OCI_Object_Registry.OP_Effective and VW_XE_OCI_Object_Registry.OP_Expires
                          And VW_XE_OCI_Object_Registry.OP_Process_YN = 'Y'
                          And VW_XE_OCI_Object_Registry.DC_Object_Process Is Not Null
                          And VW_XE_OCI_Object_Registry.DC_FILE_Direction = v_DC_FILE_Direction
                          And VW_XE_OCI_Object_Registry.OP_Current_Database_CID = v_Current_Database_CID)
        -- Does object exist in bucket?
        v_Cloud_Object_Info := Pkg_XE_OCI_Object_Registry.Qry_Cloud_Object_Info(  p_XE_OCI_Object_Registry_ID =>  c_Object_Registry.XE_OCI_Object_Registry_ID);
         Case c_Object_Registry.DC_FILE_Direction
         When 'IMPORT' Then
            If v_Cloud_Object_Info.t_Object_Name Is Not Null then
              -- Object exists in Bucket.
              -- Insert Contents into appropriate table, and move object to processed bucket.
              v_Batch_ID      := Pkg_XE_OCI_Object_Registry.Import_Object_Contents( p_XE_OCI_Object_Registry_ID =>  c_Object_Registry.XE_OCI_Object_Registry_ID,
                                                                                            p_Error_On_No_Find          =>  'Y');
              v_Processed_YN      := 'Y';
              -- No Object file to import, so skip
            End If;
          When 'EXPORT' Then
            -- Export Contents into appropriate Object.
            Pkg_XE_OCI_Object_Registry.Export_File_Contents( p_XE_OCI_Object_Registry_ID =>  c_Object_Registry.XE_OCI_Object_Registry_ID,
                                                             p_Error_On_No_Find          =>  'Y');
            v_Processed_YN      := 'Y';
             -- Either does not apply, No action necessary, or process not created for type yet.
          End Case;
        If v_Processed_YN = 'Y' Then
          -- Log in XE_OCI_Objects_Log
          v_XE_OCI_Objects_Log.XE_OCI_OBJECT_REGISTRY_ID := v_VW_XE_OCI_Object_Registry.XE_OCI_Object_Registry_ID;
          v_XE_OCI_Objects_Log.LO_OBJECT_NAME := c_Object_Registry.OP_Object_Name;
          v_XE_OCI_Objects_Log.LO_CHECKSUM := v_Cloud_Object_Info.t_CheckSum;
          v_XE_OCI_Objects_Log.LO_DATA_ROW_ACTION_CID := Pkg_Codes_Value.Qry_Code_Value_To_ID( p_CO_Name_Abrev => 'DATA_ROW_ACTION',
                                                                      p_CV_Value_Character    =>    'IMPORT');
          v_XE_OCI_Objects_Log.LO_DATA_ROW_STATUS_CID    := Pkg_Codes_Value.Qry_Code_Value_To_ID(p_CO_Name_Abrev => 'DATA_ROW_STATUS',
                                                                   p_CV_Value_Character    =>    'PROCESSED');
          v_XE_OCI_Objects_Log.LO_XE_BUCKETS_ID_CURRENT  := v_VW_XE_OCI_Object_Registry.OP_XE_Buckets_ID_Move_To;
          v_XE_OCI_Objects_Log.LO_OBJECT_CREATED := Nvl(v_Cloud_Object_Info.t_Created,v_Cloud_Object_Info.t_Last_Modified);
          v_XE_OCI_Objects_Log.LO_BATCH_ID := v_Batch_ID;
          v_XE_OCI_Objects_Log.LO_FILE_EXTENSION := c_Object_Registry.OP_File_Extension;
          v_XE_OCI_Objects_Log.LO_FILE_SIZE  := v_Cloud_Object_Info.t_Bytes;
          v_XE_OCI_Objects_Log.LO_DATE_OBJECT_PROCESSED  := Current_Date;
          v_XE_OCI_Objects_Log.LO_EFFECTIVE  := Nvl(v_Cloud_Object_Info.t_Created,v_Cloud_Object_Info.t_Last_Modified);
          v_XE_OCI_Objects_Log.LO_EXPIRES     := Pkg_Codes_Value.g_Date_Infinity;
          -- Managed by Trigger and Virtual Column
          -- v_XE_OCI_Objects_Log.SY_USER_ID_ADD            := ;
          -- v_XE_OCI_Objects_Log.SY_DATE_ADD               := ;
          -- v_XE_OCI_Objects_Log.SY_USER_ID_MOD            := ;
          -- v_XE_OCI_Objects_Log.SY_DATE_MOD               := ;
          -- v_XE_OCI_Objects_Log.VC_LO_OBJECT_NAME_CAPS    := ;
          -- v_XE_OCI_Objects_Log.VC_LO_FILE_EXTNESION_CAPS := ;
          Pkg_XE_OCI_Objects_Log.Ins( p_XE_OCI_Objects_Log  =>  v_XE_OCI_Objects_Log);
        End If;
      End Loop;
        When e_Error Then
      -- Your error handling and logging here
        When Others Then
      -- Your error handling and logging here
  End Process_Objects;

Recognizing your code will naturally be different, perhaps very different, the detail of the packages, procedures, and functions will not be shown. Instead, here are some notes on the major components:

  • Direction: Parameter value determines direction of process, import or export. Allows segmentation of processing so automated import and export processing can be scheduled and handled differently.
  • Identify Current Database Instance Code Value: Current DB Instance ID is compared to each Object Registry row so only rows associated with the current instance are processed, i.e. Dev, Test, and Prod. Current_DB_Instance is a user-defined function (Returns select ora_database_Name from dual;) returning the current DB.
  • Process Loop: Loops through each Object Registry View row to determine if it pertains to the current database instance, OP_Process_YN is (Y)es and the effective/expire range is current.
  • Object Exists in Bucket? If importing Object to be acted upon exists in expected Bucket (XE_Buckets_ID) process it if OP_OverWrite = 'Y', otherwise proceed to next Object.
  • Log: Create and Insert Log entry if a file was processed.

And some snippets for the more generic key code in the called procedures.

Get Object information from the Bucket and determine if Object exists upon return.

  v_Obj       Pkg_XE_OCI_Object_Registry.t_Cloud_Object;
  v_VW_XE_OCI_Object_Registry VW_XE_OCI_Object_Registry%RowType;
  v_VW_XE_OCI_Object_Registry := Pkg_XE_OCI_Object_Registry.Pkg_XE_OCI_Object_Registry.Qry_View( p_XE_OCI_Object_Registry_ID => c_Object_Registry.XE_OCI_Object_Registry_ID);

  SELECT Object_Name, Bytes, Checksum, Created, Last_Modified
    Into v_Obj.t_Object_Name, v_Obj.t_Bytes, v_Obj.t_CheckSum, v_Obj.t_Created, v_Obj.t_Last_Modified
    FROM DBMS_CLOUD.LIST_OBJECTS( Credential_Name => v_VW_XE_OCI_Object_Registry.BU_Credential,
                                Location_URI    => v_VW_XE_OCI_Object_Registry.BU_Full_URI)
         Where Upper(Object_Name) = Upper(APEX_Util.URL_Encode(v_VW_XE_OCI_Object_Registry.OP_Object_Name));
   Return v_Obj;  

Import Contents by creating an external table on the particular Object.

  • If external table already exists delete it to accommodate any potential file or table definition changes that might have occurred.
  • Create the new table using Object Registry view information.
  • Import and process data in Object based on newly created external table. The process importing from the external table is not shown because it is process specific and would not apply in general.
  • The external table name is retrieved separately from the Object Registry even though the next procedure retrieves the full Object Registry information for the Object for two reasons. First, it allows the external name to be defined/customized in one place so if the naming convention changes in the future, it only needs to change in this function. Second, it returns a single value that could be needed separately elsewhere without returning the whole row.
    Pkg_XE_OCI_Object_Registry.Del_External_Process_Table(  p_XE_OCI_Object_Registry_ID => p_XE_OCI_Object_Registry_ID);

    -- Retrieve external table name for this object 
    v_External_Table_Name := Pkg_XE_OCI_Object_Registry.Qry_External_Table_Name( p_XE_OCI_Object_Registry_ID =>  p_XE_OCI_Object_Registry_ID);
    -- Get Object Registry information
    v_VW_Object_Registry  := Pkg_XE_OCI_Object_Registry.Qry_View( p_XE_OCI_Object_Registry_ID => p_XE_OCI_Object_Registry_ID);

    Case v_VW_Object_Registry.DC_Object_Process
    When 'IMP_MARKET_CSV' Then
        -- Create External Table linked to Object in Bucket
          table_name => v_External_Table_Name.Ext_Table_Name,
          credential_name => v_VW_Object_Registry.BU_Credential,
          file_uri_list => v_VW_Object_Registry.OP_Full_Object_URI,
          column_list => 
                              'COL001    Varchar2(300 CHAR),
                               COL002    Varchar2(300 CHAR),
                               COL003    Varchar2(300 CHAR),
                               COL004    Varchar2(300 CHAR),
                               COL005    Varchar2(300 CHAR),
                               COL006    Varchar2(300 CHAR),
                               COL007    Varchar2(300 CHAR),
                               COL008    Varchar2(300 CHAR),
                               COL009    Varchar2(300 CHAR),
                               COL010    Varchar2(300 CHAR),
                               COL011    Varchar2(300 CHAR),
                               COL012    Varchar2(300 CHAR),
                               COL013    Varchar2(300 CHAR),
                               COL014    Varchar2(300 CHAR),
                               COL015    Varchar2(300 CHAR),
                               COL016    Varchar2(300 CHAR),
                               COL017    Varchar2(300 CHAR),
                               COL018    Varchar2(300 CHAR),
                               COL019    Varchar2(300 CHAR),
                               COL020    Varchar2(300 CHAR),
                               COL021    Varchar2(300 CHAR),
                               COL022    Varchar2(300 CHAR),
                               COL023    Varchar2(300 CHAR), 
                               COL024    Varchar2(300 CHAR)',
          format          => json_object('type' value 'csv', 'skipheaders' value '1')


After Importing Object contents move the Object to a "Processed" Bucket so it is not re-processed the next time the Import procedure runs. (Moving the Object after export is the responsibility of the sFTP server CLI script when it gets the contents and FTPs them to the destination.)

    v_Object_Contents_Blob  := DBMS_CLOUD.GET_OBJECT(
                                                    credential_name      => v_VW_XE_OCI_Object_Registry.BU_Credential,
                                                    object_uri           => v_VW_XE_OCI_Object_Registry.OP_Full_Object_URI,
                                                    startoffset          => 0,
                                                    endoffset            => 0,
                                                    compression          => Null

       credential_name      => v_VW_XE_OCI_Object_Registry.BU_Credential,
       object_uri           => v_VW_XE_OCI_Object_Registry.OP_Full_Object_URI_Move_To,
       contents             => v_Object_Contents_Blob);

       credential_name      => v_VW_XE_OCI_Object_Registry.BU_Credential,
       object_uri           => v_VW_XE_OCI_Object_Registry.OP_Full_Object_URI);

Exporting is similar. Loop through Object Registry, identify Objects for Export, and instead of importing from an external table, create Objects in their designated Buckets.

For example:

  v_Instance := Current_DB_Instance;
  v_VW_XE_OCI_Object_Registry := Pkg_XE_OCI_Object_Registry.Qry_View_Object_Name(  
                                      p_OP_Object_Name      =>  'FILE_TO_BE_SENT.xlsx',
                                      p_DC_Current_Database =>  v_Instance,
                                      p_DC_File_Direction   =>  'EXPORT');
  v_SQL  := 'Select * from ' || v_VW_XE_OCI_Object_Registry.OP_Table_Exported_From;

  l_context := apex_exec.open_query_context(
        p_location    => apex_exec.c_location_local_db,
        p_sql_query   => v_SQL );
  l_export := apex_data_export.export (
                        p_context   => l_context,
                        p_format    => apex_data_export.c_format_xlsx );
  apex_exec.close( l_context );
     credential_name => v_VW_XE_OCI_Object_Registry.BU_Credential,
     object_uri => v_VW_XE_OCI_Object_Registry.OP_Full_Object_URI,
     contents => l_export.content_blob

A CLI script running on the sFTP server checks for Objects in the ftp_outbound Bucket and FTPs them to their destination as appropriate. After they are sent the files are moved to the ftp_processed Bucket, completing the overall process.

Following is a sample de-identified Perl ftp server script that my system admin prepared that I would not have been able to do...

# This script checks to see if there are outbound files ready for sftp transfer
# to [ftp-site].com.  The outbound buckets are:
# ftp_outbound, ftp_outbound_dev, and ftp_outbound_test
# If there are files, then this script will sftp the bucket files to
# [ftp site].com.  Then after the files are successfully sftp'd
# then the files will be transferred to the processed buckets which are:
# ftp_processed, ftp_processed_dev, and ftp_processed_test
# This script needs to run as a user that is configured to be
# able to run oci command line commands.

# Define parameters
$bucket_name = "";
@file_dir = ("ftpdev", "ftptest", "ftpprod");
$oci_file_out = "/tmp/ftp_xfer.exit";
$oci_dir_xfer = "/data/ftp_xfer";

$ftp_folder_dev = "/fl_home/arriving/Devobjects";
$ftp_folder_prod = "/fl_home/arriving";
$ftp_folder_test = "/fl_home/arriving/TestObjects";

# Initialize variables
$date = scalar localtime;
$file_date = `date +_%b-%d-%Y_%H:%M`;
chomp ($file_date);
chop ($hostname = `/bin/hostname`);
$mailcmd = "| /usr/sbin/sendmail -oi -t";
#$mail_to="FileProcess\, tester\";
#$mail_to="tester2\, tester\";

# Make sure the data directories exist
foreach $env_dir (@file_dir) {
  $return_code = system ("/bin/mkdir -p $oci_dir_xfer/$env_dir");
  if ($return_code) { &mail_abort ("Cannot create the data directories in $oci_dir_xfer/$env_dir: $!\n"); }
next; }

# Make sure the data directories have the proper file permissions
$return_code = system ("/bin/chown -R testname:testname $oci_dir_xfer");
if ($return_code) { &mail_abort ("Cannot modify the ownership on the data directories in $oci_dir_xfer: $!\n"); }
$return_code = system ("/bin/chmod -R 700 $oci_dir_xfer");
if ($return_code) { &mail_abort ("Cannot modify the permissions on the data directories in $oci_dir_xfer: $!\n"); }

# Delete any files in the data directories that were not previously deleted
# for any reason
foreach $env_dir (@file_dir) {
  $return_code = system ("/bin/rm -f $oci_dir_xfer/$env_dir/*");
  if ($return_code) { &mail_abort ("Cannot delete the files in $oci_dir_xfer/$env_dir: $!\n"); }
next; }

# Download all files in all buckets
foreach $env_dir (@file_dir) {
  if ($env_dir eq "prod") { $bucket_name = "ftp_outbound"; }
  else { $bucket_name = "ftp_outbound_$env_dir"; }
  $return_code = system ("/bin/oci os object bulk-download --bucket-name $bucket_name --download-dir $oci_dir_xfer/$env_dir --no-overwrite > $oci_file_out 2>&1");
  if ($return_code) { &mail_abort ("Cannot download the files to $oci_dir_xfer/$env_dir from bucket $bucket_name: $!\n"); }
next; }

# Search all directories for any downloaded files
foreach $env_dir (@file_dir) {
  if ($env_dir eq "ftpprod") {
    $src_bucket_name = "ftp_outbound";
    $dst_bucket_name = "ftp_processed"; }
  else {
    $src_bucket_name = "ftp_outbound_$env_dir";
    $dst_bucket_name = "ftp_processed_$env_dir"; }
  opendir(BUCKET_DIR,"$oci_dir_xfer/$env_dir") or &mail_abort ("Cannot open the bucket directory $oci_dir_xfer/$env_dir: $!");
  @bucket_files = readdir(BUCKET_DIR) or &mail_abort ("Unable to read directory $oci_dir_xfer/$env_dir: $!");
  closedir (BUCKET_DIR);
  foreach $bucket_file (@bucket_files) {
    next if ($bucket_file eq ".");  # Skip the current directory entry
    next if ($bucket_file eq ".."); # Skip the parent  directory entry
    # sftp the file to
    # $return_code = system ("/bin/sftp $username_arriving_dir/$sftp_file $remote_sftp_user\@$remote_sftp_server:$remote_sftp_dir/$username/$remote_rtp_dir/$sftp_file > $oci_file_out");
    # if ($return_code) { &mail_abort ("Cannot copy the file $username_arriving_dir/$sftp_file to $username_rtp_dir/$sftp_file: $!\n"); }
    # If the file was sftp'd successfully above then copy the file to
    # the processed bucket.
    $return_code = system ("/bin/oci os object copy --bucket-name $src_bucket_name --destination-bucket $dst_bucket_name --source-object-name $bucket_file > $oci_file_out 2>&1");
    if ($return_code) { &mail_abort ("Cannot copy $bucket_file from bucket $src_bucket_name to bucket $dst_bucket_name: $!\n"); }
    # If the file was copied successfully above then delete the file from
    # the outbound bucket.
    $return_code = system ("/bin/oci os object delete --bucket-name $src_bucket_name --object-name $bucket_file --force >> $oci_file_out 2>&1");
    if ($return_code) { &mail_abort ("Cannot delete $bucket_file from bucket $src_bucket_name: $!\n"); }
    unlink "$oci_dir_xfer/$env_dir/$bucket_file" or &mail_abort ("Cannot delete $oci_dir_xfer/$env_dir/$bucket_file: $!\n");
  next; # Goto the next file in the bucket's directory
next; # Goto the next bucket's download directory

exit (0);

sub mail_notify {
  # Mail the appropriate people that a file has been received.
  my $file_name = shift;

  open (MAIL, $mailcmd);
From: sftp\@$hostname
To: $mail_to
Subject: Originator File Received: $file_name
Date: $date

$file_name has been transferred to the sftp server.
  close MAIL;

sub mail_abort {
  # Mail the appropriate person that there was an error and then abort
  # processing the files.
  my $errormsg = shift;

  open (MAIL, $mailcmd);
From: sftp\@$hostname
To: $mail_to
Subject: $0 on $hostname aborted processing!
Date: $date

$0 on $hostname on $date had a severe error!\n
  close MAIL;
  exit (1);

In summary, the scalable, versatile, data driven architecture helps deliver increased productivity, consistency, flexibility, and insight. While everyone's processes are different the intent is to offer a framework and ideas you can build on.

I hope you find this information useful. Comments (and likes :) ) welcome !