title: Apache Airflow - connecting to AWS S3 error
body: <p>I'm trying to get S3 hook in Apache Airflow using the Connection object.</p>
  <p>It looks like this:</p>
  <pre><code>class S3ConnectionHandler:
      def __init__():
          # values are read from configuration class, which loads from env. variables
          self._s3 = Connection(
              conn_type=&quot;s3&quot;,
              conn_id=config.AWS_CONN_ID,
              login=config.AWS_ACCESS_KEY_ID,
              password=config.AWS_SECRET_ACCESS_KEY,
              extra=json.dumps({&quot;region_name&quot;: config.AWS_DEFAULT_REGION}),
          )
  
      @property
      def s3(self) -&gt; Connection:
          return get_live_connection(self.logger, self._s3)
  
      @property
      def s3_hook(self) -&gt; S3Hook:
          return self.s3.get_hook()
  </code></pre>
  <p>I get an error:</p>
  <pre><code>Broken DAG: [...] Traceback (most recent call last):
    File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/models/connection.py&quot;, line 282, in get_hook
      return hook_class(**{conn_id_param: self.conn_id})
    File &quot;/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/amazon/aws/hooks/base_aws.py&quot;, line 354, in __init__
      raise AirflowException('Either client_type or resource_type must be provided.')
  airflow.exceptions.AirflowException: Either client_type or resource_type must be provided.
  </code></pre>
  <p>Why does this happen? From what I understand the S3Hook calls the constructor from the parent class, AwsHook, and passes the client_type as &quot;s3&quot; string. How can I fix this?</p>
  <p>I took this configuration for hook from <a href="https://stackoverflow.com/questions/66014947/creating-boto3-s3-client-on-airflow-with-an-s3-connection-and-s3-hook">here</a>.</p>
  <p><strong>EDIT:</strong> I even get the same error when directly creating the S3 hook:</p>
  <pre><code>    @property
      def s3_hook(self) -&gt; S3Hook:
          #return self.s3.get_hook()
          return S3Hook(
              aws_conn_id=config.AWS_CONN_ID,
              region_name=self.config.AWS_DEFAULT_REGION,
              client_type=&quot;s3&quot;,
              config={&quot;aws_access_key_id&quot;: self.config.AWS_ACCESS_KEY_ID, &quot;aws_secret_access_key&quot;: self.config.AWS_SECRET_ACCESS_KEY}
          )
  ``
  </code></pre>