To add Oracle to the profile mapping in Cosmos, you can follow the structure of the existing profile mappings. Here are the steps you can take:

1. Create the Oracle Profile Mapping Class:

   • Create a new directory for Oracle under cosmos/profiles/oracle/.
   • Implement the Oracle profile mapping class in a new file, e.g., cosmos/profiles/oracle/user_pass.py.

   Example implementation:

   from cosmos.profiles.base import BaseProfileMapping
   
   class OracleUserPasswordProfileMapping(BaseProfileMapping):
       @property
       def profile(self) -> dict:
           return {
               'type': 'oracle',
               'user': self.connection.login,
               'password': self.connection.password,
               'host': self.connection.host,
               'port': self.connection.port,
               'dbname': self.connection.schema,
           }

2. Update the __init__.py File:

   • In cosmos/profiles/oracle/__init__.py, import the new profile mapping class.

   from .user_pass import OracleUserPasswordProfileMapping
   
   __all__ = ["OracleUserPasswordProfileMapping"]

3. Register the Oracle Profile Mapping:

   • Add the Oracle profile mapping to the profile_mappings list in cosmos/profiles/__init__.py.

   from .oracle.user_pass import OracleUserPasswordProfileMapping
   
   profile_mappings: list[Type[BaseProfileMapping]] = [
       # ... other mappings ...
       OracleUserPasswordProfileMapping,
   ]

4. Write Unit Tests:

   • Create unit tests for the Oracle profile mapping in a new file, e.g., tests/profiles/oracle/test_oracle_user_password.py.

   Example test:

   import unittest
   from cosmos.profiles.oracle.user_pass import OracleUserPasswordProfileMapping
   from airflow.models import Connection
   
   class TestOracleUserPasswordProfileMapping(unittest.TestCase):
       def test_profile_mapping(self):
           connection = Connection(
               conn_id='oracle_default',
               conn_type='oracle',
               host='localhost',
               login='user',
               password='password',
               schema='dbname',
               port=1521
           )
           mapping = OracleUserPasswordProfileMapping(connection)
           profile = mapping.profile
           self.assertEqual(profile['type'], 'oracle')
           self.assertEqual(profile['user'], 'user')
           self.assertEqual(profile['password'], 'password')
           self.assertEqual(profile['host'], 'localhost')
           self.assertEqual(profile['port'], 1521)
           self.assertEqual(profile['dbname'], 'dbname')
   
   if __name__ == '__main__':
       unittest.main()

By following these steps, you can implement and test the Oracle profile mapping in Cosmos.

To continue talking to Dosu, mention @dosu.