[Feature] Add oracle to profile mapping

[slords]

Description

Oracle is a supported type in both airflow and cosmos. It would be nice to have the profile mapping implemented.

Use case/motivation

Map oracle from airflow to cosmos

Related issues

No response

Are you willing to submit a PR?

• [X] Yes, I am willing to submit a PR!

[dosubot[bot]]

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.

[github-actions[bot]]

This issue is stale because it has been open for 30 days with no activity.