Repository: manoss96/fluke (language: python)

Issue:
30 create filter param in methods Dir.transfer to



Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f26dc7e..0f6c18b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,47 @@
 # Change Log
 All notable changes to this project will be documented in this file.
 
+## [0.4.0] - 2023/07/24
+
+### Added
+
+- Dir API transfer_to method now receives a filter param in
+  order to filter out files during the transfer.
+  (https://github.com/manoss96/fluke/issues/30)
+
+- There now exists a module *fluke.queues* which contains a
+  number of classes used in order to access message queues
+  provided by various message queue services in the cloud.
+  (https://github.com/manoss96/fluke/issues/31)
+
+- Added a new dependency: ``azure-storage-queue==12.6.0``
+  (https://github.com/manoss96/fluke/issues/31)
+
+### Changed
+
+- *fluke.auth.AzureAuth* no longer has a constructor
+  for creating an *AzureAuth* instance via an Azure
+  service principal. Instead, this is now achieved
+  via an *AzureAuth.from_service_principal* method.
+  (https://github.com/manoss96/fluke/issues/31)
+
+- Classes *fluke.storage.{AWSS3File, AWSS3Dir}*
+  have been renamed to *fluke.storage.{AmazonS3File, AmazonS3Dir}*.
+  (https://github.com/manoss96/fluke/issues/34)
+
+- Updated dependency versions:
+  + azure-identity: 1.12.0 -> 1.13.0
+  + azure-storage-blob: 12.15.0 -> 12.17.0
+  + boto3: 1.26.84 -> 1.28.0
+  + paramiko: 3.0.0 -> 3.2.0
+
+### Fixed
+
+- Fixed issue where type of objects could not be inferred
+  when they were instantiate via a context manager.
+  (https://github.com/manoss96/fluke/issues/35)
+
+
 ## [0.3.0] - 2023/04/16
 
 ### Added
@@ -90,17 +131,17 @@ All notable changes to this project will be documented in this file.
   with the directory that spawned them. In the case of remote files, they
   also share the same client and cache. (https://github.com/manoss96/fluke/issues/16)
 
-- *fluke.storage.AWSS3File* now has a *get_bucket_name* method.
+- *fluke.storage.AmazonS3File* now has a *get_bucket_name* method.
   (https://github.com/manoss96/fluke/issues/16)
    
 ### Changed
 
-- Parameter *path* in *fluke.storage.AWSS3Dir* and *fluke.storage.AzureBlobDir*
+- Parameter *path* in *fluke.storage.AmazonS3Dir* and *fluke.storage.AzureBlobDir*
   constructors now defaults to ``None`` in order to reference the whole
   container and bucket respectively (https://github.com/manoss96/fluke/issues/15)
 
 - Providing a path that starts with a separator, e.g. ``/``, to constructors
-  *fluke.storage.AWSS3Dir* and *fluke.storage.AzureBlobDir* will now throw
+  *fluke.storage.AmazonS3Dir* and *fluke.storage.AzureBlobDir* will now throw
   an ``InvalidPathError`` exception. (https://github.com/manoss96/fluke/issues/15)
 
 - Both *File* and *Dir* API methods *get_metadata* will now return
diff --git a/README.md b/README.md
index 968da32..5b02d2e 100644
--- a/README.md
+++ b/README.md
@@ -8,15 +8,27 @@
 <!-- What is Fluke? -->
 ## What is Fluke?
 
-Fluke is a Python package that is primarily to be used as a data transfer tool.
-By utilizing Fluke, moving your data between two remote locations can be
-done in just a matter of seconds from the comfort of your own machine!
+Fluke is a Python package that is to be used as a higher-level API to
+cloud services that primarily relate to object storage and message queues.
+Fluke manages to hide away much of the complexity that derives from working
+with said services, aiding you in completing your tasks fast and hassle-free!
+Fluke achieves this by:
+
+* Treating object storage services as traditional file systems,
+  unifying the two under a single *File/Dir* API, through which
+  you are able to manage your data no matter where they reside,
+  be it a local/remote file system or a bucket in the cloud.
+
+* Greatly reducing the intricacies of interacting with message queues
+  by thinking of them as mere data structures that support three elementary
+  operations, that is, push/peek/poll.
 
 
 <!-- Installation -->
 ## Installation
 
-You can start using Fluke by installing it via pip. Note that *fluke* requires Python >= 3.9.
+You can start using Fluke by installing it via pip.
+Note that *fluke* requires Python >= 3.9.
 
 ```sh
 pip install fluke-api
@@ -26,14 +38,24 @@ pip install fluke-api
 <!-- Usage example -->
 ## Usage Example
 
-In this example, we are going to transfer an entire directory residing
-within a remote machine to the cloud, more specifically, to an Amazon S3 bucket.
+In this example, we will be using Fluke in order to:
 
-First things first, we need to be able to authenticate with both the remote
-machine and AWS. In order to achieve this, we will be importing from ``fluke.auth``:
+1. Poll an Amazon SQS queue every minute for new messages. Each of these messages contains the
+   path of a newly uploaded file to an Amazon S3 bucket.
+2. Use the content of these messages in order to locate and access said files within the bucket.
+3. If a file's metadata field ``transfer`` has been set to ``True``, then transfer it to a remote server.
+
+First things first, we need to be able to authenticate with both AWS
+and the remote server. In order to achieve this, we will be importing from ``fluke.auth``:
 
 ```python
-from fluke.auth import RemoteAuth, AWSAuth
+from fluke.auth import AWSAuth, RemoteAuth
+
+# This object will be used to authenticate
+# with AWS.
+aws_auth = AWSAuth(
+    aws_access_key_id="aws_access_key",
+    aws_secret_access_key="aws_secret_key")
 
 # This object will be used to authenticate
 # with the remote machine.
@@ -41,25 +63,28 @@ rmt_auth = RemoteAuth.from_password(
     hostname="host",
     username="user",
     password="password")
-
-# This object will be used to authenticate
-# with AWS.
-aws_auth = AWSAuth(
-    aws_access_key_id="aws_access_key",
-    aws_secret_access_key="aws_secret_key")
 ```
 
-Next, we just need to import from ``fluke.storage`` so that we
-gain access to any necessary resources and perform the data transfer:
+Next, we just need to import from ``fluke.queues`` and ``fluke.storage``
+so that we gain access to any necessary resources in order to perform
+the data transfer:
 
 ```python
-from fluke.storage import RemoteDir, AWSS3Dir
+from fluke.queues import AmazonSQSQueue
+from fluke.storage import AmazonS3Dir, RemoteDir
 
 with (
-    RemoteDir(auth=rmt_auth, path='/home/user/dir') as rmt_dir,
-    AWSS3Dir(auth=aws_auth, bucket="bucket", path='dir', create_if_missing=True) as aws_dir
+    AmazonSQSQueue(auth=aws_auth, queue='queue') as queue,
+    AmazonS3Dir(auth=aws_auth, bucket='bucket') as bucket,
+    RemoteDir(auth=rmt_auth, path='/home/user/dir', create_if_missing=True) as rmt_dir
 ):
-    rmt_dir.transfer_to(dst=aws_dir, recursively=True)
+    for batch in queue.poll(polling_frequency=60):
+        for msg in batch:
+            file = bucket.get_file(path=msg)
+            file.load_metadata()
+            metadata = file.get_metadata()
+            if bool(metadata['transfer']):
+                file.transfer_to(dst=rmt_dir)
 ```
 
 And that's basically it!
diff --git a/docs/source/conf.py b/docs/source/conf.py
index 05f2273..185cd5d 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -22,7 +22,7 @@
 author = 'Manos Stoumpos'
 
 # The full version, including alpha/beta/rc tags
-release = '0.3.0'
+release = '0.4.0'
 
 
 # -- General configuration ---------------------------------------------------
diff --git a/docs/source/documentation/queues.rst b/docs/source/documentation/queues.rst
new file mode 100644
index 0000000..dd49de4
--- /dev/null
+++ b/docs/source/documentation/queues.rst
@@ -0,0 +1,7 @@
+***********************
+fluke.queues
+***********************
+.. automodule:: fluke.queues
+   :members:
+   :undoc-members:
+   :inherited-members:
\ No newline at end of file
diff --git a/docs/source/getting_started.rst b/docs/source/getting_started.rst
index 53b0086..cda39c8 100644
--- a/docs/source/getting_started.rst
+++ b/docs/source/getting_started.rst
@@ -4,9 +4,20 @@
 What is Fluke?
 *******************
 
-Fluke is a Python package that is primarily to be used as a data transfer tool.
-By utilizing Fluke, moving your data between two remote locations can be
-done in just a matter of seconds from the comfort of your own machine!
+Fluke is a Python package that is to be used as a higher-level API to
+cloud services that primarily relate to object storage and message queues.
+Fluke manages to hide away much of the complexity that derives from working
+with said services, aiding you in completing your tasks fast and hassle-free!
+Fluke achieves this by:
+
+* Treating object storage services as traditional file systems,
+  unifying the two under a single *File/Dir* API, through which
+  you are able to manage your data no matter where they reside,
+  be it a local/remote file system or a bucket in the cloud.
+
+* Greatly reducing the intricacies of interacting with message queues
+  by thinking of them as mere data structures that support three elementary
+  operations, that is, push/peek/poll.
 
 *******************
 Installation
@@ -19,28 +30,25 @@ Note that *fluke* requires Python >= 3.9.
 
     pip install fluke-api
 
+
 *******************
 Usage example
 *******************
 
+In this example, we will be using Fluke in order to:
 
-In this example, we are going to transfer an entire directory residing
-within a remote machine to the cloud, more specifically, to an Amazon S3 bucket.
+1. Poll an Amazon SQS queue every minute for new messages. Each of these messages contains the
+   path of a newly uploaded file to an Amazon S3 bucket.
+2. Use the content of these messages in order to locate and access said files within the bucket.
+3. If a file's metadata field ``transfer`` has been set to ``True``, then transfer it to a remote server.
 
-First things first, we need to be able to authenticate with both the remote
-machine and AWS. In order to achieve this, we will be importing from
+First things first, we need to be able to authenticate with both AWS
+and the remote server. In order to achieve this, we will be importing from
 `fluke.auth <documentation/auth.html>`_:
 
 .. code-block:: python
 
-  from fluke.auth import RemoteAuth, AWSAuth
-
-  # This object will be used to authenticate
-  # with the remote machine.
-  rmt_auth = RemoteAuth.from_password(
-      hostname="host",
-      username="user",
-      password="password")
+  from fluke.auth import AWSAuth, RemoteAuth
 
   # This object will be used to authenticate
   # with AWS.
@@ -48,19 +56,35 @@ machine and AWS. In order to achieve this, we will be importing from
       aws_access_key_id="aws_access_key",
       aws_secret_access_key="aws_secret_key")
 
+  # This object will be used to authenticate
+  # with the remote server.
+  rmt_auth = RemoteAuth.from_password(
+      hostname="host",
+      username="user",
+      password="password")
+
 
-Next, we just need to import from `fluke.storage <documentation/storage.html>`_
-so that we gain access to any necessary resources and perform the data transfer:
+Next, we just need to import from `fluke.queues <documentation/queues.html>`_
+and `fluke.storage <documentation/storage.html>`_ so that we gain access to any
+necessary resources in order to perform the data transfer:
 
 .. code-block:: python
 
-  from fluke.storage import RemoteDir, AWSS3Dir
+  from fluke.queues import AmazonSQSQueue
+  from fluke.storage import AmazonS3Dir, RemoteDir
 
   with (
-      RemoteDir(auth=rmt_auth, path='/home/user/dir') as rmt_dir,
-      AWSS3Dir(auth=aws_auth, bucket="bucket", path='dir', create_if_missing=True) as aws_dir
+      AmazonSQSQueue(auth=aws_auth, queue='queue') as queue,
+      AmazonS3Dir(auth=aws_auth, bucket='bucket') as bucket,
+      RemoteDir(auth=rmt_auth, path='/home/user/dir', create_if_missing=True) as rmt_dir
   ):
-      rmt_dir.transfer_to(dst=aws_dir, recursively=True)
+    for batch in queue.poll(polling_frequency=60):
+        for msg in batch:
+            file = bucket.get_file(path=msg)
+            file.load_metadata()
+            metadata = file.get_metadata()
+            if bool(metadata['transfer']):
+                file.transfer_to(dst=rmt_dir)
 
 And that's basically it!
 
diff --git a/docs/source/index.rst b/docs/source/index.rst
index 61e3862..92bf636 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -23,8 +23,9 @@ visiting fluke on `Github <https://github.com/manoss96/fluke>`_.
    :maxdepth: 1
    :caption: User Guide
 
-   user_guide/handling_auth
-   user_guide/interacting_with_data
+   user_guide/authentication
+   user_guide/file_and_object_storage
+   user_guide/message_queues
 
 
 .. toctree::
@@ -33,3 +34,4 @@ visiting fluke on `Github <https://github.com/manoss96/fluke>`_.
 
    documentation/auth
    documentation/storage
+   documentation/queues
diff --git a/docs/source/user_guide/handling_auth.rst b/docs/source/user_guide/authentication.rst
similarity index 84%
rename from docs/source/user_guide/handling_auth.rst
rename to docs/source/user_guide/authentication.rst
index 9fdbba7..2193331 100644
--- a/docs/source/user_guide/handling_auth.rst
+++ b/docs/source/user_guide/authentication.rst
@@ -1,7 +1,7 @@
 .. _ug_authentication:
 
 ***********************
-Handling authentication
+Authentication
 ***********************
 
 Authentication is the process by which a user verifies their identity. All remote
@@ -9,20 +9,26 @@ resources require some form of authentication during which a user must identify
 themselves so that they are granted access to said resources. Fluke itself
 manages authentication through the `fluke.auth <../documentation/auth.html>`_ module.
 The general idea is that you must first initialize some *Auth* instance,
-which is then provided to the resource with which you wish to interact:
+which is then provided to the resource with which you wish to interact.
+For instance, in the code snippet below we use the
+`RemoteAuth <../documentation/auth.html#fluke.auth.RemoteAuth>`_
+class in order to access a file that resides within a remote server.
 
 .. code-block:: python
 
-  from fluke.auth import SomeAuthClass
+  from fluke.auth import RemoteAuth
+  from fluke.storage import RemoteFile
 
-  # Create an "Auth" class instance.
-  auth = SomeAuthClass(**credentials)
+  # Create a "RemoteAuth" class instance.
+  auth = RemoteAuth.from_password(
+      hostname='host',
+      username='user',
+      password='pwd')
 
   # Provide it to the resource.
-  resource = SomeResourceClass(auth=auth, **params)
-
-  # You can now interact with the resource!
-  # ...
+  with RemoteFile(auth=auth, path='/home/user/file.txt') as file:
+    # You can now interact with the resource!
+    file_bytes = file.read()
 
 Below, we are going to take a look at all different types of authentication.
 
@@ -87,17 +93,17 @@ Authenticating with Azure
 ==========================================
 
 Authentication with Azure resources can happen
-in two ways: either with an Azure service principal
-or with a connection string. Below, we'll take a
-look at both.
+in two ways: either with an Azure Active Directory
+(AD) service principal or with a connection string.
+Below, we'll take a look at both.
 
-----------------------------------------------
-Authenticating via an Azure service principal
-----------------------------------------------
+---------------------------------------
+Authenticating via a service principal
+---------------------------------------
 
-In order to authenticate via an Azure service principal,
-you may use the ordinary `AzureAuth <../documentation/auth.html#fluke.auth.AzureAuth>`_
-class constructor, providing it with the storage account's URL, as well as
+In order to authenticate via a service principal, you may use the
+`AzureAuth.from_service_principal <../documentation/auth.html#fluke.auth.AzureAuth.from_service_principal>`_
+method, providing it with the storage account's URL, as well as
 all information relevant to the service principal, that is,
 the *tenant ID*, *client ID* and finally the *client secret*:
 
@@ -105,7 +111,7 @@ the *tenant ID*, *client ID* and finally the *client secret*:
 
   from fluke.auth import AzureAuth
 
-  auth = AzureAuth(
+  auth = AzureAuth.from_service_principal(
       account_url = 'https://ACCOUNT.blob.core.windows.net'
       tenant_id = 'TENANT_ID',
       client_id = 'CLIENT_ID',
@@ -119,10 +125,10 @@ Authenticating via a connection string
 
 Alternatively, you are able to use
 `AzureAuth.from_conn_string <../documentation/auth.html#fluke.auth.AzureAuth.from_conn_string>`_
-so as to authenticate with a resource by using a connection string,
+so as to gain access to a specific set of Azure resources by using a connection string,
 which must typically include the account's name and key, as well
-as all necessary endpoints regarding the resources to which we request
-access to:
+as all necessary endpoints regarding the resources to which you
+request access to:
 
 .. code-block:: python
 
diff --git a/docs/source/user_guide/interacting_with_data.rst b/docs/source/user_guide/file_and_object_storage.rst
similarity index 86%
rename from docs/source/user_guide/interacting_with_data.rst
rename to docs/source/user_guide/file_and_object_storage.rst
index 4fc0942..2076e96 100644
--- a/docs/source/user_guide/interacting_with_data.rst
+++ b/docs/source/user_guide/file_and_object_storage.rst
@@ -1,33 +1,36 @@
 .. _ug_storage:
 
 ***********************
-Interacting with data
+File & Object Storage
 ***********************
 
 Fluke makes it easy to interact with your data no matter
-where they currently reside or where they are going! In
-this section, we will look at all aspects of working with
-data through Fluke.
+where they reside! In this section, we will look at all
+aspects of working with data through Fluke.
 
 ========================================
-Acquiring access to data
+Accessing data
 ========================================
 
-Fluke offers two distinct APIs in order to interact with data:
+Fluke treats object storage in the cloud as if it were an
+ordinary file system where files are organized in directories,
+which is the type of storage most people are familiar with.
+Therefore, Fluke offers two distinct, yet interconnected, APIs
+in order to interact with data:
 
 #. **File API**: Used for handling either individual files in a file system,
    be it local or remote, or objects in the cloud. This API includes:
    
      * `LocalFile <../documentation/storage.html#fluke.storage.LocalFile>`_
      * `RemoteFile <../documentation/storage.html#fluke.storage.RemoteFile>`_
-     * `AWSS3File <../documentation/storage.html#fluke.storage.AWSS3File>`_
+     * `AmazonS3File <../documentation/storage.html#fluke.storage.AmazonS3File>`_
      * `AzureBlobFile <../documentation/storage.html#fluke.storage.AzureBlobFile>`_
 
 #. **Dir API**: Used for handling entire directories of files/objects. This API includes:
 
      * `LocalDir <../documentation/storage.html#fluke.storage.LocalDir>`_
      * `RemoteDir <../documentation/storage.html#fluke.storage.RemoteDir>`_
-     * `AWSS3Dir <../documentation/storage.html#fluke.storage.AWSS3Dir>`_
+     * `AmazonS3Dir <../documentation/storage.html#fluke.storage.AmazonS3Dir>`_
      * `AzureBlobDir <../documentation/storage.html#fluke.storage.AzureBlobDir>`_
 
 ----------------------------------------
@@ -51,7 +54,7 @@ Accessing remote entities
 For any entity that is not local to our machine's file system, we usually
 need a bit more information. This is typically an *Auth* instance used
 for authenticating with the remote host in order to gain access to its resources
-(see `Handling authentication <authentication.html>`_), as well as some
+(see `Authentication <authentication.html>`_), as well as some
 more information regarding an entity's exact location. For example, in the
 case of a directory which resides within an Amazon S3 bucket, we also need
 the name of said bucket:
@@ -59,22 +62,22 @@ the name of said bucket:
 .. code-block:: python
 
   from fluke.auth import AWSAuth
-  from fluke.storage import AWSS3Dir
+  from fluke.storage import AmazonS3Dir
 
   # Create an "AWSAuth" instance.
   auth = AWSAuth(**credentials)
 
   # Gain access to an Amazon S3 directory.
-  aws_dir = AWSS3Dir(
+  s3_dir = AmazonS3Dir(
     auth=auth,
     bucket='bucket_name',
     path='path/within/amazon/s3/bucket/dir')
 
-  # Interact with "aws_dir".
+  # Interact with "s3_dir".
   ...
 
   # Close any open connections.
-  aws_dir.close()
+  s3_dir.close()
 
 Notice how at the end we invoked the instance's ``close`` method? This is
 done in order to prevent any connection leaks from occurring, and is in fact
@@ -85,19 +88,19 @@ class's context manager:
 .. code-block:: python
 
   from fluke.auth import AWSAuth
-  from fluke.storage import AWSS3Dir
+  from fluke.storage import AmazonS3Dir
 
   # Create an "AWSAuth" instance.
   auth = AWSAuth(**credentials)
 
   # Gain access to an S3 directory through a context manager.
-  with AWSS3Dir(
+  with AmazonS3Dir(
     auth=auth,
     bucket='bucket_name',
     path='path/within/amazon/s3/bucket/dir'
-  ) as aws_dir:
+  ) as s3_dir:
 
-    # Interact with "aws_dir".
+    # Interact with "s3_dir".
     ...
 
 .. _intermediate-access:
@@ -153,23 +156,23 @@ directory:
 .. code-block:: python
 
   from fluke.auth import AWSAuth
-  from fluke.storage import AWSS3Dir, AWSS3File
+  from fluke.storage import AmazonS3Dir, AmazonS3File
 
   # Create an "AWSAuth" instance.
   auth = AWSAuth(**credentials)
 
   # Access directory.
-  aws_dir = AWSS3Dir(auth=auth, bucket='bucket_name', path='dir')
+  s3_dir = AmazonS3Dir(auth=auth, bucket='bucket_name', path='dir')
 
   # Access file through directory.
-  aws_file: AWSS3File = aws_dir.get_file('file.txt')
+  s3_file: AmazonS3File = s3_dir.get_file('file.txt')
 
   # Close connection through file.
-  aws_file.close()
+  s3_file.close()
 
   # This would now cause an exception as
   # the underlying connection has been closed.
-  dir_size = aws_dir.get_size()
+  dir_size = s3_dir.get_size()
 
 Therefore, by utilizing intermediate access, you are able to
 interact with many remote entities without having to open
@@ -178,25 +181,25 @@ multiple connections:
 .. code-block:: python
 
   from fluke.auth import AWSAuth
-  from fluke.storage import AWSS3Dir, AWSS3File
+  from fluke.storage import AmazonS3Dir, AmazonS3File
 
   # Create an "AWSAuth" instance.
   auth = AWSAuth(**credentials)
 
   # Gain access to the entire bucket.
-  with AWSS3Dir(auth=auth, bucket='bucket_name') as bucket:
+  with AmazonS3Dir(auth=auth, bucket='bucket_name') as bucket:
 
     # Interact with its files.
-    aws_file: AWSS3File = bucket.get_file('file.txt')
-    file_size = aws_file.get_size()
+    s3_file: AmazonS3File = bucket.get_file('file.txt')
+    file_size = s3_file.get_size()
     
     # Interact with its subdirectories.
-    aws_dir: AWSS3Dir = bucket.get_subdir('dir')
-    aws_dir.ls()
+    s3_dir: AmazonS3Dir = bucket.get_subdir('dir')
+    s3_dir.ls()
 
 
 ========================================
-Exploring data
+Interacting with data
 ========================================
 
 After having gained access to an entity, you are then
@@ -330,19 +333,19 @@ all in just a few lines of code:
 .. code-block:: python
 
   from fluke.auth import AWSAuth, AzureAuth
-  from fluke.storage import AWSS3Dir, AzureBlobDir
+  from fluke.storage import AmazonS3Dir, AzureBlobDir
 
   # This object will be used to authenticate with AWS.
   aws_auth = AWSAuth(**aws_credentials)
 
   # This object will be used to authenticate with Azure.
-  azr_auth = AzureAuth(**azr_credentials)
+  azr_auth = AzureAuth.from_service_principal(**azr_credentials)
 
   with (
-      AWSS3Dir(auth=aws_auth, bucket='bucket', path='dir') as aws_dir,
+      AmazonS3Dir(auth=aws_auth, bucket='bucket', path='dir') as s3_dir,
       AzureBlobDir(auth=azr_auth, container='container', path='dir') as azr_dir
   ):
-      aws_dir.transfer_to(dst=azr_dir, recursively=True)
+      s3_dir.transfer_to(dst=azr_dir, recursively=True)
 
 Unless you set parameter ``suppress_output`` to ``True``, Fluke will go
 on to print the progress of the transfer onto the console:
@@ -393,15 +396,15 @@ some metadata to it through the ``set_metadata`` method:
 .. code-block:: python
 
   from fluke.auth import AWSAuth
-  from fluke.storage import LocalFile, AWSS3Dir
+  from fluke.storage import LocalFile, AmazonS3Dir
 
   # Instantiate a local file and assign some metadata to it.
   file = LocalFile(path='/home/user/path/to/file.txt')
   file.set_metadata({'id': '12345', 'type': 'txt'})
 
   # Transfer file to Amazon S3 along with its metadata.
-  with AWSS3Dir(auth=AWSAuth(**aws_credentials), bucket='bucket', path='dir') as aws_dir:
-      file.transfer_to(dst=aws_dir, include_metadata=True)
+  with AmazonS3Dir(auth=AWSAuth(**aws_credentials), bucket='bucket', path='dir') as s3_dir:
+      file.transfer_to(dst=s3_dir, include_metadata=True)
 
 Along with *file.txt* being uploaded to Amazon S3, any metadata that
 were defined via the ``set_metadata`` method will be associated with it.
@@ -409,7 +412,7 @@ In fact, we can easily confirm this by executing the following code:
 
 .. code-block:: python
 
-  print(aws_dir.get_metadata('file.txt'))
+  print(s3_dir.get_metadata('file.txt'))
 
 which results in the following output being printed onto the console:
 
@@ -484,10 +487,10 @@ assigned to it. Let's do just that and see what happens:
 .. code-block:: python
 
   from fluke.auth import AWSAuth
-  from fluke.storage import AWSS3File
+  from fluke.storage import AmazonS3File
 
   # Gain access to 'file.txt' on Amazon S3 and print its metadata.
-  with AWSS3File(auth=AWSAuth(**aws_credentials), bucket='bucket', path='dir/file.txt') as aws_obj:
+  with AmazonS3File(auth=AWSAuth(**aws_credentials), bucket='bucket', path='dir/file.txt') as aws_obj:
       print(aws_obj.get_metadata())
 
 By executing the above code, we get the following output:
@@ -515,10 +518,10 @@ add the aforementioned line of code:
 .. code-block:: python
 
   from fluke.auth import AWSAuth
-  from fluke.storage import AWSS3File
+  from fluke.storage import AmazonS3File
 
   # Gain access to 'file.txt' on Amazon S3 and print its metadata.
-  with AWSS3File(auth=AWSAuth(**aws_credentials), bucket='bucket', path='dir/file.txt') as aws_obj:
+  with AmazonS3File(auth=AWSAuth(**aws_credentials), bucket='bucket', path='dir/file.txt') as aws_obj:
       # Load metadata first.
       aws_obj.load_metadata()
       # Then print it.
@@ -542,18 +545,18 @@ has not been invoked:
 .. code-block:: python
 
   from fluke.auth import AWSAuth, AzureAuth
-  from fluke.storage import AWSS3File, AzureBlobFile
+  from fluke.storage import AmazonS3File, AzureBlobFile
 
   with (
-      AWSS3File(auth=AWSAuth(**aws_credentials), bucket='bucket', path='dir/file.txt') as aws_obj,
-      AzureBlobDir(auth=AzureAuth(**azr_credentials), container='container', path='file.txt') as azr_dir
+      AmazonS3File(auth=AWSAuth(**aws_credentials), bucket='bucket', path='dir/file.txt') as aws_obj,
+      AzureBlobDir(auth=AzureAuth.from_service_principal(**azr_credentials), container='container', path='file.txt') as azr_dir
   ):
       aws_obj.transfer_to(dst=azr_dir, include_metadata=True)
 
 ..  _speeding-things-up-with-caching:
 
 ========================================
-Speeding things up with caching
+Catching data
 ========================================
 
 Whenever you request some piece of information about a remote entity,
@@ -579,19 +582,19 @@ setting parameter ``cache`` to ``True`` during its instantiation:
   import time
   
   from fluke.auth import AWSAuth
-  from fluke.storage import AWSS3File
+  from fluke.storage import AmazonS3File
 
   auth = AWSAuth(**aws_credentials)
 
-  with AWSS3Dir(auth=auth, bucket='bucket', path='dir', cache=True) as aws_dir:
+  with AmazonS3Dir(auth=auth, bucket='bucket', path='dir', cache=True) as s3_dir:
     # Fetch metadata via HTTP.
     t = time.time()
-    aws_dir.load_metadata()
+    s3_dir.load_metadata()
     print(f"Fetched metadata in {time.time() - t:.2f} seconds!")
 
     # Fetch metadata from cache.
     t = time.time()
-    aws_dir.load_metadata()
+    s3_dir.load_metadata()
     print(f"Fetched metadata in {time.time() - t:.2f} seconds!")
 
 Executing the above code block outputs the following:
@@ -610,23 +613,23 @@ by invoking ``purge``:
 .. code-block:: python
   
   from fluke.auth import AWSAuth
-  from fluke.storage import AWSS3File
+  from fluke.storage import AmazonS3File
 
-  with AWSS3Dir(auth=AWSAuth(**aws_credentials), bucket='bucket', path='dir', cache=True) as aws_dir:
+  with AmazonS3Dir(auth=AWSAuth(**aws_credentials), bucket='bucket', path='dir', cache=True) as s3_dir:
     # Count number of items in directory.
-    print(f"Directory count: {aws_dir.count()}")
+    print(f"Directory count: {s3_dir.count()}")
 
     # At this point, assume that one more
     # file is uploaded to the directory.
 
     # Re-count number of items in directory
     # without purging the cache.
-    print(f"Directory count: {aws_dir.count()}")
+    print(f"Directory count: {s3_dir.count()}")
 
     # Re-count number of items in directory
     # after purging the cache.
-    aws_dir.purge()
-    print(f"Directory count: {aws_dir.count()}")
+    s3_dir.purge()
+    print(f"Directory count: {s3_dir.count()}")
 
 By executing the above code, we get the following output:
 
@@ -645,28 +648,28 @@ them. Consider the following example:
 
 .. code-block:: python
 
-  from fluke.storage import AWSS3Dir
+  from fluke.storage import AmazonS3Dir
 
   # This object will be used to authenticate with AWS.
   aws_auth = AWSAuth(**aws_credentials)
 
   # Access an AWS S3 directory and render it "cacheable".
-  with AWSS3Dir(auth=aws_auth, bucket="bucket", path='dir', cache=True) as aws_dir:
+  with AmazonS3Dir(auth=aws_auth, bucket="bucket", path='dir', cache=True) as s3_dir:
     # Fetch the directory's total size and time it.
     t = time.time()
-    _ = aws_dir.get_size(recursively=True)
+    _ = s3_dir.get_size(recursively=True)
     print(f"Fetched size in {time.time() - t:.2f} seconds!")
 
     # Now purge the directory's cache.
-    aws_dir.purge()
+    s3_dir.purge()
 
     # Fetch the sizes of the directory's file and subdirectory.
-    _ = aws_dir.get_file('file.txt').get_size()
-    _ = aws_dir.get_subdir('subdir').get_size(recursively=True)
+    _ = s3_dir.get_file('file.txt').get_size()
+    _ = s3_dir.get_subdir('subdir').get_size(recursively=True)
 
     # Fetch the directory's total size and time it again.
     t = time.time()
-    _ = aws_dir.get_size(recursively=True)
+    _ = s3_dir.get_size(recursively=True)
     print(f"Fetched size in {time.time() - t:.2f} seconds!")
 
 The above code produces the following output when executed:
diff --git a/docs/source/user_guide/message_queues.rst b/docs/source/user_guide/message_queues.rst
new file mode 100644
index 0000000..5fda80f
--- /dev/null
+++ b/docs/source/user_guide/message_queues.rst
@@ -0,0 +1,255 @@
+.. _ug_queues:
+
+***********************
+Message Queues
+***********************
+
+Through the module `fluke.queues <../documentation/queues.html>`_ Fluke
+provides a straightforward API that makes it easy to interact with various
+message queue services in the cloud, such as Amazon SQS and Azure Queue
+Storage. The simplicity of this API comes from the fact that it is based
+on just three basic operations:
+
+1. Push - Sending messages to a queue.
+2. Peek - Getting a glimpse of the messages within a queue.
+3. Poll - Fetching messages from a queue, thereby deleting them.
+
+Below we'll take a closer look on each one.
+
+===============================
+Pushing messages into a queue
+===============================
+
+As already mentioned, the *push* operation is used in order
+to send messages to the queue. After gaining access to a queue,
+you can simply invoke its ``push`` method, providing it with
+the string message you wish to send to the queue:
+
+.. code-block:: python
+
+  from fluke.auth import AWSAuth
+  from fluke.queues import AmazonSQSQueue
+
+  auth = AWSAuth(
+      aws_access_key_id="aws_access_key",
+      aws_secret_access_key="aws_secret_key")
+
+  with AmazonSQSQueue(auth=auth, queue='queue') as queue:
+    is_delivered = queue.push(message="Hello!")
+
+In response to invoking the ``push`` method, a boolean value
+is returned which indicates whether the message has been
+successfully delivered to the queue or not.
+
+===============================
+Peeking at messages in a queue
+===============================
+
+Next up, you might like to take a peek at the messages in a
+queue without actually deleting them. This operation can come
+quite handy when you want for example to observe the format of the
+messages within a queue so that you are able to build a parser in
+order to extract the information you are interested in. Whatever its use,
+peeking at messages is possible by invoking a queue instance's ``peek``
+method, which goes on to return a list that contains up to 10 messages
+found within the queue:
+
+.. code-block:: python
+
+  from fluke.auth import AWSAuth
+  from fluke.queues import AmazonSQSQueue
+
+  auth = AWSAuth(
+      aws_access_key_id="aws_access_key",
+      aws_secret_access_key="aws_secret_key")
+
+  with AmazonSQSQueue(auth=auth, queue='queue') as queue:
+    messages = queue.peek()
+
+.. warning::
+
+    Please note that even though the ``peek`` method does not explicitly
+    delete messages, it can still result in a message being removed from
+    the queue, as every time we fetch a message its *receive count* is
+    increased by one. Most queues are configured in such a way so that if
+    a message is fetched a specific number of times, then it is automatically
+    deleted.
+
+===============================
+Polling messages from a queue
+===============================
+
+Finally, you want to be able to fetch messages from a queue
+while at the same time removing them from it entirely. This
+is where the *poll* operation comes in. By invoking a queue
+instance's ``poll`` method, you essentially get access to an
+iterator that is capable of going through the queue's messages
+in distinct batches, removing any received messages in the process
+of doing so:
+
+.. code-block:: python
+
+  from fluke.auth import AWSAuth
+  from fluke.queues import AmazonSQSQueue
+
+  auth = AWSAuth(
+      aws_access_key_id="aws_access_key",
+      aws_secret_access_key="aws_secret_key")
+
+  with AmazonSQSQueue(auth=auth, queue='queue') as queue:
+    for batch in queue.poll(batch_size=10):
+        for msg in batch:
+            # Process the message...
+
+By default, the ``poll`` method will go on to fetch all of the messages
+available in the queue. If you just want to fetch a particular
+number of messages, you can set the method's ``num_messages``
+parameter accordingly:
+
+.. code-block:: python
+
+  from fluke.auth import AWSAuth
+  from fluke.queues import AmazonSQSQueue
+
+  auth = AWSAuth(
+      aws_access_key_id="aws_access_key",
+      aws_secret_access_key="aws_secret_key")
+
+  with AmazonSQSQueue(auth=auth, queue='queue') as queue:
+    for batch in queue.poll(num_messages=100, batch_size=10):
+        for msg in batch:
+            # Process the message...
+
+
+-----------------------------
+Continuous polling
+-----------------------------
+
+When either all messages or a subset of them has been successfully received,
+depending on the value of ``num_messages``, the iterator reaches its ending
+point and the program carries on. If you wish to keep on polling the queue for
+any new messages, then this is possible by setting the ``polling_frequency``
+parameter:
+
+.. code-block:: python
+
+  from fluke.auth import AWSAuth
+  from fluke.queues import AmazonSQSQueue
+
+  auth = AWSAuth(
+      aws_access_key_id="aws_access_key",
+      aws_secret_access_key="aws_secret_key")
+
+  with AmazonSQSQueue(auth=auth, queue='queue') as queue:
+    for batch in queue.poll(
+        num_messages=100,
+        batch_size=10,
+        polling_frequency=60
+    ):
+        for msg in batch:
+            # Process the message...
+
+By defining a value for this parameter, each time the specified number
+of messages has been fetched from the queue, the program will wait for
+a total number of seconds equal to the value of ``polling_frequency``,
+at which point the queue is polled again until either a total of ``num_messages``
+messages are received or the queue is empty. This goes on indefinitely until
+either an unhandled exception occurs or the process is explicitly killed.
+
+-----------------------------
+Pre- vs post-delivery delete
+-----------------------------
+
+Whenever a batch of messages is delivered, the messages within
+the batch have not yet actually been deleted from the queue,
+and will only be deleted just before the next batch of messages arrives,
+after all messages within the current batch have already been processed.
+This is the queue's default behaviour when polling messages, as defined
+by the default value of parameter ``pre_delivery_delete``, and it guarantees
+that no batch of messages will be lost in case something goes wrong during
+their processing:
+
+.. code-block:: python
+
+  from fluke.auth import AWSAuth
+  from fluke.queues import AmazonSQSQueue
+
+  auth = AWSAuth(
+      aws_access_key_id="aws_access_key",
+      aws_secret_access_key="aws_secret_key")
+
+  with AmazonSQSQueue(auth=auth, queue='queue') as queue:
+    for batch in queue.poll():
+        # Although an exception is raised, no messages will be lost
+        # as, at this point, the messages within the batch have yet
+        # to be deleted.
+        raise Exception()
+
+Nevertheless, if something goes wrong after we have already processed
+one or more messages within the batch, then there is always the danger
+of reprocessing the exact same messages as they may be included in any
+one of the subsequent batches:
+
+.. code-block:: python
+
+  from fluke.auth import AWSAuth
+  from fluke.queues import AmazonSQSQueue
+
+  auth = AWSAuth(
+      aws_access_key_id="aws_access_key",
+      aws_secret_access_key="aws_secret_key")
+
+  with AmazonSQSQueue(auth=auth, queue='queue') as queue:
+    for batch in queue.poll(num_messages=2, batch_size=2):
+        for i, msg in enumerate(batch):
+            if i == 0:
+                # Process message...
+            else:
+                raise Exception()
+
+In the example above, the first message of the batch will be processed
+as expected. However, during the processing of the second message, an
+exception will be thrown, causing the program to exit. This means that
+despite the second message being the only one that was not processed,
+the queue still contains both messages, as the next batch of messages
+never arrived.
+
+If you want to avoid this, you can set ``pre_delivery_delete`` to ``True``.
+This results in any batch of messages being removed from the queue before
+they are actually delivered to you:
+
+.. code-block:: python
+
+  from fluke.auth import AWSAuth
+  from fluke.queues import AmazonSQSQueue
+
+  auth = AWSAuth(
+      aws_access_key_id="aws_access_key",
+      aws_secret_access_key="aws_secret_key")
+
+  with AmazonSQSQueue(auth=auth, queue='queue') as queue:
+    for batch in queue.poll(pre_delivery_delete=True):
+        # At this point, all messages within the batch
+        # have already been removed from the queue.
+
+As a general rule of thumb, set ``pre_delivery_delete`` to ``False``
+if you don't mind processing the same message twice and are more concerned
+in losing a message without processing it first, whereas set ``pre_delivery_delete``
+to ``True`` if you cannot afford to process the same message more than once.
+
+.. warning::
+
+    Pre- and post-delivery are concepts related to Fluke and do not
+    apply to the underlying message queue services. Note that you
+    should always enforce your own rules in order to check for duplicate
+    messages if this is a matter of concern to you, as such messages
+    can still arrive regardless of the value of ``pre_delivery_delete``.
+    For instance, the default *standard* queues offered by Amazon SQS
+    support *at-least-once delivery* which dictates that any messages
+    pushed to the queue will be delivered to you at least once, but there
+    is always the chance that a message be re-delivered due to the distributed
+    nature of the queues. In order to avoid any suprises, be sure to read the
+    specifications of the type of queue you are working with and take any
+    necessary precautions, no matter whether you are accessing said queue
+    through its standard API or through the API provided by Fluke.
+
diff --git a/fluke/_handlers.py b/fluke/_handlers.py
index 9bbd52f..3634b93 100644
--- a/fluke/_handlers.py
+++ b/fluke/_handlers.py
@@ -21,8 +21,8 @@
 from ._iohandlers import LocalFileWriter as _LocalFileWriter
 from ._iohandlers import RemoteFileReader as _RemoteFileReader
 from ._iohandlers import RemoteFileWriter as _RemoteFileWriter
-from ._iohandlers import AWSS3FileReader as _AWSS3FileReader
-from ._iohandlers import AWSS3FileWriter as _AWSS3FileWriter
+from ._iohandlers import AmazonS3FileReader as _AmazonS3FileReader
+from ._iohandlers import AmazonS3FileWriter as _AmazonS3FileWriter
 from ._iohandlers import AzureBlobReader as _AzureBlobReader
 from ._iohandlers import AzureBlobWriter as _AzureBlobWriter
 from ._exceptions import UnknownKeyTypeError as _UKTE
@@ -961,16 +961,16 @@ def mkdir(self, path: str) -> None:
             ContentType='application/x-directory; charset=UTF-8')
         
 
-    def get_reader(self, file_path: str) -> _AWSS3FileReader:
+    def get_reader(self, file_path: str) -> _AmazonS3FileReader:
         '''
-        Returns an ``AWSS3FileReader`` class instance \
+        Returns an ``AmazonS3FileReader`` class instance \
         used for reading from a file which resides within \
         an Amazon S3 bucket.
 
         :param str file_path: The absolute path of \
             the file in question.
         '''
-        return _AWSS3FileReader(
+        return _AmazonS3FileReader(
             file_path=file_path,
             file_size=self.get_file_size(file_path),
             bucket=self.__bucket)
@@ -981,9 +981,9 @@ def get_writer(
         file_path: str,
         metadata: _Optional[dict[str, str]],
         in_chunks: bool
-    ) -> _AWSS3FileWriter:
+    ) -> _AmazonS3FileWriter:
         '''
-        Returns an ``AWSS3FileWriter`` class instance \
+        Returns an ``AmazonS3FileWriter`` class instance \
         used for writing to a file which resides within \
         an Amazon S3 bucket.
 
@@ -997,7 +997,7 @@ def get_writer(
             write the file in distinct chunks or \
             all at once.
         '''
-        return _AWSS3FileWriter(
+        return _AmazonS3FileWriter(
             file_path=file_path,
             metadata=metadata,
             in_chunks=in_chunks,
diff --git a/fluke/_iohandlers.py b/fluke/_iohandlers.py
index d7ffe30..2f6531c 100644
--- a/fluke/_iohandlers.py
+++ b/fluke/_iohandlers.py
@@ -512,7 +512,7 @@ def _write_impl(self, chunk: bytes) -> int:
         return len(chunk)
     
 
-class AWSS3FileReader(_FileReader):
+class AmazonS3FileReader(_FileReader):
     '''
     A class used in reading from files which \
     reside within an Amazon S3 bucket.
@@ -573,7 +573,7 @@ def _read_impl(
         return self.__file.get(Range=range)['Body'].read()
 
             
-class AWSS3FileWriter(_FileWriter):
+class AmazonS3FileWriter(_FileWriter):
     '''
     A class used in writing to files which \
     reside within an Amazon S3 bucket.
diff --git a/fluke/auth.py b/fluke/auth.py
index f2537e4..b6facf8 100644
--- a/fluke/auth.py
+++ b/fluke/auth.py
@@ -213,29 +213,28 @@ def get_credentials(self) -> dict[str, str]:
 class AzureAuth():
     '''
     This class is used for authenticating with Microsoft Azure.
-
-    :param str account_url: The URI to the storage account.
-    :param str tenant_id: ID of the service principal's tenant.
-    :param str client_id: The service principal's client ID.
-    :param str client_secret: One of the service principal's client secrets.
     '''
 
-    def __init__(
-        self,
+    @classmethod
+    def from_service_principal(
+        cls,
         account_url: str,
         tenant_id: str,
         client_id: str,
         client_secret: str
-    ):
+    ) -> 'AzureAuth':
         '''
-        This class is used for authenticating with Microsoft Azure.
+        Returns an ``AzureAuth`` instance used for \
+        authenticating with Microsoft Azure via a, \
+        Azure service principal.
 
         :param str account_url: The URI to the storage account.
         :param str tenant_id: ID of the service principal's tenant.
         :param str client_id: The service principal's client ID.
         :param str client_secret: One of the service principal's client secrets.
         '''
-        self.__credentials = {
+        auth = cls()
+        auth.__credentials = {
             'account_url': account_url,
             'tenant_id': tenant_id,
             'client_id': client_id,
@@ -246,9 +245,10 @@ def __init__(
                 pattern=r"https?://([^.]+).blob.core.windows.net/?",
                 string=account_url)
             if match is not None:
-                self.__storage_account = match.group(1)
+                auth.__storage_account = match.group(1)
         else:
-            self.__storage_account = None
+            auth.__storage_account = None
+        return auth
 
 
     @classmethod
@@ -258,7 +258,7 @@ def from_conn_string(cls, conn_string: str) -> 'AzureAuth':
         authenticating with Microsoft Azure via a \
         connection string.
         '''
-        auth = cls(None, None, None, None)
+        auth = cls()
         auth.__credentials = {
             'conn_string': conn_string
         }
diff --git a/fluke/queues.py b/fluke/queues.py
new file mode 100644
index 0000000..203d9b0
--- /dev/null
+++ b/fluke/queues.py
@@ -0,0 +1,710 @@
+import time as _time
+import random as _rand
+import warnings as _warn
+from abc import ABC as _ABC
+from abc import abstractmethod as _absmethod
+from typing import Iterator as _Iterator
+from typing import Optional as _Optional
+
+
+import boto3 as _boto3
+from azure.identity import ClientSecretCredential as _CSC
+from azure.storage.queue import QueueClient as _QueueClient
+
+
+from .auth import AWSAuth as _AWSAuth
+from .auth import AzureAuth as _AzureAuth
+
+
+class _Queue(_ABC):
+    '''
+    An abstract class which serves as the \
+    base class for all queue classes.
+
+    :param str name: The name of the queue.
+    '''
+
+    def __init__(self, name: str) -> None:
+        '''
+        An abstract class which serves as the \
+        base class for all queue classes.
+
+        :param str name: The name of the queue.
+        '''
+        self.__name = name
+        self.open()
+
+
+    def __enter__(self) -> '_Queue':
+        '''
+        Enter the runtime context related to this instance.
+        '''
+        return self
+
+
+    def __exit__(self, exc_type, exc_value, traceback) -> None:
+        '''
+        Exit the runtime context related to this object. 
+        '''
+        self.close()
+
+
+    def __del__(self) -> None:
+        '''
+        The class destructor method.
+        '''
+        if self.is_open():
+            # Display warning.
+            msg = f'You might want to consider instantiating class "{self.__class__.__name__}"'
+            msg += " through the use of a context manager by utilizing Python's"
+            msg += ' "with" statement, or by simply invoking an instance\'s'
+            msg += ' "close" method after being done using it.'
+            _warn.warn(msg, ResourceWarning)
+            # Close connections.
+            self.close()
+
+
+    def get_name(self) -> str:
+        '''
+        Returns the name of the queue to which \
+        a connection has been established.
+        '''
+        return self.__name
+
+
+    @_absmethod
+    def is_open(self) -> bool:
+        '''
+        Returns a value indicating whether \
+        this handler's underlying client connection \
+        is open or not.
+        '''
+        pass
+
+
+    @_absmethod
+    def open(self) -> None:
+        '''
+        Opens all necessary connections.
+        '''
+        pass
+
+
+    @_absmethod
+    def close(self) -> None:
+        '''
+        Close all open connections.
+        '''
+        pass
+
+
+    @_absmethod
+    def count(self) -> int:
+        '''
+        Returns the total number of messages that \
+        are residing within the queue at the time \
+        of the request.
+        '''
+        pass
+
+
+    @_absmethod
+    def push(
+        self,
+        message: str,
+        suppress_output: bool = False
+    ) -> bool:
+        '''
+        Pushes the provided message into the queue. \
+        Returns ``True`` if said message was successfully \
+        pushed into the queue, else returns ``False``.
+
+        :param str message: A string message.
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+        '''
+        pass
+
+
+    @_absmethod
+    def peek(self, suppress_output: bool = False) -> list[str]:
+        '''
+        Returns a list containing at most ten messages \
+        currently residing within the queue.
+
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+
+        :note: This method does not go on to explicitly \
+            remove messages from the queue. However, any \
+            messages returned by this method will have their \
+            "receive count" increased, which in turn might \
+            result in said messages being removed from the \
+            queue in case the queue's maximum receive count \
+            threshold is exceeded.
+        '''
+        pass
+
+
+    @_absmethod
+    def poll(
+        self,
+        num_messages: _Optional[int] = None,
+        batch_size: int = 10,
+        polling_frequency: _Optional[int] = None,
+        pre_delivery_delete: bool = False,
+        suppress_output: bool = False
+    ) -> _Iterator[list[str]]:
+        '''
+        Iterates through the messages available in the queue \
+        in distinct batches, deleting them in the process of \
+        doing so.
+
+        :param int | None num_messages: The number of messages to \
+            iterate through. If set to ``None``, then the queue \
+            is constantly querried for new messages until there \
+            are none left. Defaults to ``None``.
+        :param int batch_size: The maximum number of messages \
+            a single batch may contain. Deafults to ``10``.
+        :param int | None polling_frequency: If set to an integer \
+            value, then the queue is going to be being polled at \
+            regular time intervals equal to said value in seconds. \
+            If set to ``None``, then the queue is only polled once. \
+            Defaults to ``None``.
+        :param bool pre_delivery_delete: Indicates whether a \
+            batch of messages is to be removed from the queue \
+            before or after its delivery. If set to ``True``, \
+            then it is guaranteed that any delivered messages will \
+            have already been removed from the queue, thus reducing \
+            the likelihood of fetching the same message twice. If set \
+            to ``False``, then any delivered messages are only deleted \
+            just before the delivery of the next batch of messages, \
+            thus preventing from any messages being lost in case \
+            an error occurs during their processing. Defaults \
+            to ``False``.
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+        '''
+        pass
+
+
+    @_absmethod
+    def clear(self, suppress_output: bool = False) -> None:
+        '''
+        Empties the queue by deleting all messages.
+
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+        '''
+        pass
+
+
+class AmazonSQSQueue(_Queue):
+    '''
+    This class represents an Amazon SQS queue.
+
+    :param AWSAuth auth: An ``AWSAuth`` instance \
+        used in authenticating with AWS.
+    :param str queue: The name of the Amazon SQS queue \
+        to which a connection is to be established.
+    '''
+
+    def __init__(self, auth: _AWSAuth, queue: str):
+        '''
+        This class represents an Amazon SQS queue.
+
+        :param AWSAuth auth: An ``AWSAuth`` instance \
+            used in authenticating with AWS.
+        :param str queue: The name of the Amazon SQS queue \
+            to which a connection is to be established.
+        '''
+        self.__auth = auth
+        self.__queue = None
+        super().__init__(name=queue)
+    
+
+    def is_open(self) -> bool:
+        '''
+        Returns a value indicating whether \
+        this handler's underlying client connection \
+        is open or not.
+        '''
+        return self.__queue is not None
+
+
+    def open(self) -> None:
+        '''
+        Opens an HTTP connection to \
+        the Amazon SQS queue.
+        '''
+
+        if self.is_open():
+            return
+        
+        # Fix deprecated endpoint.
+        creds = self.__auth.get_credentials()
+        if (region := creds['region_name']) is not None:
+            creds.update({'endpoint_url': f"https://sqs.{region}.amazonaws.com"})
+
+        print(f"\nEstablishing connection to '{self.get_name()}' Amazon SQS queue...")
+        self.__queue = _boto3.resource(
+            service_name='sqs', **creds
+        ).get_queue_by_name(QueueName=self.get_name())
+        print("Connection established.")
+
+
+    def close(self):
+        '''
+        Closes the HTTP connection to \
+        the Amazon SQS queue.
+        '''
+        if self.__queue is not None:
+            self.__queue.meta.client.close()
+            self.__queue = None
+
+
+    def count(self) -> int:
+        '''
+        Returns the total number of messages that \
+        are residing within the queue at the time \
+        of the request.
+        '''
+        self.__queue.reload()
+        return (
+            int(self.__queue.attributes['ApproximateNumberOfMessages']) +
+            int(self.__queue.attributes['ApproximateNumberOfMessagesNotVisible'])
+        )
+
+
+    def push(
+        self,
+        message: str,
+        suppress_output: bool = False
+    ) -> bool:
+        '''
+        Pushes the provided message into the queue. \
+        Returns ``True`` if said message was successfully \
+        pushed into the queue, else returns ``False``.
+
+        :param str message: A string message.
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+        '''
+        if not suppress_output:
+            print(f'\nPushing message "{message}" into queue "{self.get_name()}".')
+        
+        try:
+            self.__queue.send_message(
+                MessageBody=message,
+                DelaySeconds=0)
+            if not suppress_output:
+                print("Message sent successfully!")
+            return True
+        except Exception as e:
+            if not suppress_output:
+                print(f"Failed to send message: {e}")
+            return False
+    
+
+    def peek(self, suppress_output: bool = False) -> list[str]:
+        '''
+        Returns a list containing at most ten messages \
+        currently residing within the queue.
+
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+
+        :note: This method does not go on to explicitly \
+            remove messages from the queue. However, any \
+            messages returned by this method will have their \
+            "receive count" increased, which in turn might \
+            result in said messages being removed from the \
+            queue in case the queue's maximum receive count \
+            threshold is exceeded.
+        '''
+        if not suppress_output:
+            print(f'\nPeeking messages in queue "{self.get_name()}".')
+
+        return [
+            msg.body for msg in self.__queue.receive_messages(
+                AttributeNames=['QueueUrl'],
+                VisibilityTimeout=1,
+                MaxNumberOfMessages=_rand.randint(1, 10))
+        ]
+
+    
+    def poll(
+        self,
+        num_messages: _Optional[int] = None,
+        batch_size: int = 10,
+        polling_frequency: _Optional[int] = None,
+        pre_delivery_delete: bool = False,
+        suppress_output: bool = False
+    ) -> _Iterator[list[str]]:
+        '''
+        Iterates through the messages available in the queue \
+        in distinct batches, deleting them in the process of \
+        doing so.
+
+        :param int | None num_messages: The number of messages to \
+            iterate through. If set to ``None``, then the queue \
+            is constantly querried for new messages until there \
+            are none left. Defaults to ``None``.
+        :param int batch_size: The maximum number of messages \
+            a single batch may contain. Deafults to ``10``.
+        :param int | None polling_frequency: If set to an integer \
+            value, then the queue is going to be being polled at \
+            regular time intervals equal to said value in seconds. \
+            If set to ``None``, then the queue is only polled once. \
+            Defaults to ``None``.
+        :param bool pre_delivery_delete: Indicates whether a \
+            batch of messages is to be removed from the queue \
+            before or after its delivery. If set to ``True``, \
+            then it is guaranteed that any delivered messages will \
+            have already been removed from the queue, thus reducing \
+            the likelihood of fetching the same message twice. If set \
+            to ``False``, then any delivered messages are only deleted \
+            just before the delivery of the next batch of messages, \
+            thus preventing from any messages being lost in case \
+            an error occurs during their processing. Defaults \
+            to ``False``.
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+        '''
+        while True:
+
+            if not suppress_output:
+                print(f'\nPolling messages from queue "{self.get_name()}".')
+
+            num_messages_fetched = 0
+
+            while num_messages is None or num_messages_fetched < num_messages:
+
+                batch = []
+                while len(batch) < batch_size:
+                    microbatch = self.__queue.receive_messages(
+                        AttributeNames=['QueueUrl'],
+                        VisibilityTimeout=30,
+                        MaxNumberOfMessages=min(
+                            batch_size - len(batch),
+                            10 if num_messages is None
+                            else num_messages-num_messages_fetched,
+                            10))
+                    
+                    if len(microbatch) == 0:
+                        break
+                    
+                    batch += microbatch
+                    num_messages_fetched += len(microbatch)
+
+                    if num_messages_fetched == num_messages:
+                        break
+
+                if len(batch) == 0:
+                    break
+                
+                entries, messages = [], []
+
+                for i in range(0, len(batch), 10):
+                    for j, msg in enumerate(batch[i:i+10]):
+                        entries.append({'Id': str(i+j), 'ReceiptHandle': msg.receipt_handle})
+                        messages.append(msg.body)
+
+                if pre_delivery_delete:
+                    # First remove messages from queue.
+                    resp = self.__queue.delete_messages(Entries=entries)
+                    if not suppress_output and 'Failed' in resp:
+                        for msg in map(lambda d: d['Message'], resp['Failed']):
+                            print(f'Failed to delete message "{msg}".')
+                    # Filter out any messages that failed to be removed.
+                    deleted_messages = []
+                    for j in map(lambda d: int(d['Id']), resp['Successful']):
+                        deleted_messages.append(messages[j])
+                    # Only deliver successfully deleted messages.
+                    yield deleted_messages
+                    num_messages_fetched += len(deleted_messages)
+                else:
+                    # First deliver messages.
+                    yield messages
+                    num_messages_fetched += len(messages)
+                    # Then attempt to remove them from queue.
+                    resp = self.__queue.delete_messages(Entries=entries)
+                    if not suppress_output and 'Failed' in resp:
+                        for msg in map(lambda d: d['Message'], resp['Failed']):
+                            print(f'Failed to delete message "{msg}".')
+
+            if polling_frequency is None:
+                break
+            else:
+                _time.sleep(polling_frequency)
+
+
+    def clear(self, suppress_output: bool = False) -> None:
+        '''
+        Empties the queue by deleting all messages.
+
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+        '''
+        if not suppress_output:
+            print(f"Deleting all messages from queue '{self.get_name()}'.")
+        self.__queue.purge()
+
+
+    def __enter__(self) -> 'AmazonSQSQueue':
+        '''
+        Enter the runtime context related to this instance.
+        '''
+        return super().__enter__()
+
+
+class AzureStorageQueue(_Queue):
+    '''
+    A class used in handling the HTTP \
+    connection to an Azure Queue Storage \
+    queue.
+
+    :param AzureAuth auth: An ``AzureAuth`` instance \
+        used in authenticating with Microsoft Azure.
+    :param str queue: The name of the Azure Queue Storage \
+        queue to which a connection is to be established.
+    '''
+
+    def __init__(self, auth: _AzureAuth, queue: str):
+        '''
+        A class used in handling the HTTP \
+        connection to an Azure Queue Storage \
+        queue.
+
+        :param AzureAuth auth: An ``AzureAuth`` instance \
+            used in authenticating with Microsoft Azure.
+        :param str queue: The name of the Azure Queue Storage \
+            queue to which a connection is to be established.
+        '''
+        self.__auth = auth
+        self.__queue = None
+        super().__init__(name=queue)
+    
+
+    def is_open(self) -> bool:
+        '''
+        Returns a value indicating whether \
+        this handler's underlying client connection \
+        is open or not.
+        '''
+        return self.__queue is not None
+
+
+    def open(self) -> None:
+        '''
+        Opens an HTTP connection to the \
+        Azure Queue Storage queue.
+        '''
+
+        if self.__queue is not None:
+            return
+
+        credentials = self.__auth.get_credentials()
+
+        print(f"\nEstablishing connection to '{self.get_name()}' Azure Queue Storage queue...")
+        if 'conn_string' in credentials:
+            self.__queue = _QueueClient.from_connection_string(
+                conn_str=credentials['conn_string'],
+                queue_name=self.get_name())
+        else:
+            self.__queue = _QueueClient(
+                account_url=credentials.pop('account_url'),
+                queue_name=self.get_name(),
+                credential=_CSC(**credentials))
+        print("Connection established.")
+
+
+    def close(self):
+        '''
+        Closes the HTTP connection to the \
+        Azure Queue Storage queue.
+        '''
+        if self.__queue is not None:
+            self.__queue.close()
+            self.__queue = None
+
+
+    def count(self) -> int:
+        '''
+        Returns the total number of messages that \
+        are residing within the queue at the time \
+        of the request.
+        '''
+        return (self.__queue
+            .get_queue_properties()
+            .approximate_message_count)
+
+
+    def push(
+        self,
+        message: str,
+        suppress_output: bool = False
+    ) -> bool:
+        '''
+        Pushes the provided message into the queue. \
+        Returns ``True`` if said message was successfully \
+        pushed into the queue, else returns ``False``.
+
+        :param str message: A string message.
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+        '''
+        if not suppress_output:
+            print(f'\nPushing message "{message}" into queue "{self.get_name()}".')
+
+        try:
+            self.__queue.send_message(content=message)
+            if not suppress_output:
+                print("Message sent successfully!")
+            return True
+        except Exception as e:
+            if not suppress_output:
+                print(f"Failed to send message: {e}")
+            return False
+    
+
+    def peek(self, suppress_output: bool = False) -> list[str]:
+        '''
+        Returns a list containing at most ten messages \
+        currently residing within the queue.
+
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+
+        :note: This method does not go on to explicitly \
+            remove messages from the queue. However, any \
+            messages returned by this method will have their \
+            "receive count" increased, which in turn might \
+            result in said messages being removed from the \
+            queue in case the queue's maximum receive count \
+            threshold is exceeded.
+        '''
+        if not suppress_output:
+            print(f'\nPeeking messages in queue "{self.get_name()}".')
+
+        return [
+            msg.content for msg in self.__queue.peek_messages(
+                max_messages=_rand.randint(1, 10))
+        ]
+
+    
+    def poll(
+        self,
+        num_messages: _Optional[int] = None,
+        batch_size: int = 10,
+        polling_frequency: _Optional[int] = None,
+        pre_delivery_delete: bool = False,
+        suppress_output: bool = False
+    ) -> _Iterator[list[str]]:
+        '''
+        Iterates through the messages available in the queue \
+        in distinct batches, deleting them in the process of \
+        doing so.
+
+        :param int | None num_messages: The number of messages to \
+            iterate through. If set to ``None``, then the queue \
+            is constantly querried for new messages until there \
+            are none left. Defaults to ``None``.
+        :param int batch_size: The maximum number of messages \
+            a single batch may contain. Deafults to ``10``.
+        :param int | None polling_frequency: If set to an integer \
+            value, then the queue is going to be being polled at \
+            regular time intervals equal to said value in seconds. \
+            If set to ``None``, then the queue is only polled once. \
+            Defaults to ``None``.
+        :param bool pre_delivery_delete: Indicates whether a \
+            batch of messages is to be removed from the queue \
+            before or after its delivery. If set to ``True``, \
+            then it is guaranteed that any delivered messages will \
+            have already been removed from the queue, thus reducing \
+            the likelihood of fetching the same message twice. If set \
+            to ``False``, then any delivered messages are only deleted \
+            just before the delivery of the next batch of messages, \
+            thus preventing from any messages being lost in case \
+            an error occurs during their processing. Defaults \
+            to ``False``.
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+        '''
+        while True:
+
+            if not suppress_output:
+                print(f'\nPolling messages from queue "{self.get_name()}".')
+
+            num_messages_fetched = 0
+
+            while num_messages is None or num_messages_fetched < num_messages:
+
+                no_messages_left = True
+
+                for batch in self.__queue.receive_messages(
+                        messages_per_page=min(
+                            batch_size,
+                            10 if num_messages is None
+                            else num_messages-num_messages_fetched,
+                            10),
+                        max_messages=
+                            None if num_messages is None
+                            else num_messages - num_messages_fetched,
+                        visibility_timeout=30
+                ).by_page():
+                    if pre_delivery_delete:
+                        # First attempt to remove messages from queue.
+                        messages = []
+                        for msg in batch:
+                            try:
+                                self.__queue.delete_message(msg.id, msg.pop_receipt)
+                                messages.append(msg.content)
+                            except:
+                                if not suppress_output:
+                                    print(f'Failed to delete message "{msg}".')
+                        # Then deliver messages.
+                        yield messages
+                        num_messages_fetched += len(messages)
+                    else:
+                        # First deliver messages.
+                        messages = list(batch)
+                        yield [msg.content for msg in messages]
+                        num_messages_fetched += len(messages)
+                        # Then attempt to remove messages from queue.
+                        for msg in messages:
+                            try:
+                                self.__queue.delete_message(msg.id, msg.pop_receipt)
+                            except:
+                                if not suppress_output:
+                                    print(f'Failed to delete message "{msg}".')
+                    # Indicate there are still messages left.
+                    no_messages_left = False
+
+                if no_messages_left:
+                    break
+
+            if polling_frequency is None:
+                break
+            else:
+                _time.sleep(polling_frequency)
+
+
+
+    def clear(self, suppress_output: bool = False) -> None:
+        '''
+        Empties the queue by deleting all messages.
+
+        :param bool suppress_output: If set to ``True``, then \
+            suppresses all output. Defaults to ``False``.
+        '''
+        if not suppress_output:
+            print(f"Deleting all messages from queue '{self.get_name()}'.")
+        self.__queue.clear_messages()
+
+
+    def __enter__(self) -> 'AzureStorageQueue':
+        '''
+        Enter the runtime context related to this instance.
+        '''
+        return super().__enter__()
\ No newline at end of file
diff --git a/fluke/storage.py b/fluke/storage.py
index 72b0b0e..f31a3d8 100644
--- a/fluke/storage.py
+++ b/fluke/storage.py
@@ -1,8 +1,8 @@
 
 
 __all__ = [
-    'AWSS3Dir',
-    'AWSS3File',
+    'AmazonS3Dir',
+    'AmazonS3File',
     'AzureBlobDir',
     'AzureBlobFile',
     'LocalFile',
@@ -353,11 +353,7 @@ def _get_handler(self) -> _ClientHandler:
         return self.__handler
     
 
-    def __new__(
-        cls,
-        *args,
-        **kwargs
-    ) -> 'LocalFile':
+    def __new__(cls, *args, **kwargs):
         '''
         Creates an instance of this class.
 
@@ -679,7 +675,7 @@ def load_metadata(self) -> None:
         self.set_metadata(metadata=metadata)
 
 
-class AWSS3File(_CloudFile):
+class AmazonS3File(_CloudFile):
     '''
     This class represents an object which resides \
     within an Amazon S3 bucket.
@@ -783,9 +779,9 @@ def _create_file(
         path: str,
         handler: _AWSClientHandler,
         metadata: dict[str, str]
-    ) -> 'AWSS3File':
+    ) -> 'AmazonS3File':
         '''
-        Creates and returns an ``AWSS3File`` instance.
+        Creates and returns an ``AmazonS3File`` instance.
 
         :param str path: The path pointing to the file.
         :param AWSClientHandler handler: An ``AWSClientHandler`` \
@@ -803,7 +799,7 @@ class instance.
         return instance
     
 
-    def __enter__(self) -> 'AWSS3File':
+    def __enter__(self) -> 'AmazonS3File':
         '''
         Enter the runtime context related to this instance.
         '''
@@ -1239,6 +1235,7 @@ def transfer_to(
         overwrite: bool = False,
         include_metadata: bool = False,
         chunk_size: _typ.Optional[int] = None,
+        filter: _typ.Optional[_typ.Callable[[str], bool]] = None,
         suppress_output: bool = False,
     ) -> bool:
         '''
@@ -1264,15 +1261,31 @@ def transfer_to(
         :param int | None chunk_size: If not ``None``, then files are \
             transferred in chunks, whose size are equal to this parameter \
             value. Defaults to ``None``.
+        :param Callable[[str], bool] | None filter: A filter function \
+            to be used in order to exclude certain files from being \
+            transferred. This function is considered to receive a single \
+            parameter, namely the file's absolute path within the source \
+            directory, and to return a boolean value, based on which it is \
+            determined whether the file is to be filtered out during the \
+            transfer (``True``) or not (``False``). Defaults to ``None``.
         :param bool suppress_output: If set to ``True``, then \
             suppresses all output. Defaults to ``False``.
         '''
+        if filter is None:
+            filter = lambda _: True
+
         # Store the directory's files in a list.
-        file_paths = [fp for fp in self.__handler.traverse_dir(
-            dir_path = self.get_path(),
-            recursively=recursively,
-            include_dirs=False,
-            show_abs_path=True)]
+        if not suppress_output:
+            print("\nListing files to be transferred...")
+        file_paths = [
+            fp for fp in self.__handler.traverse_dir(
+                dir_path = self.get_path(),
+                recursively=recursively,
+                include_dirs=False,
+                show_abs_path=True)
+            if filter(fp)]
+        if not suppress_output:
+            print("Listing operation completed.")
 
         total_num_files = len(file_paths)
         failures = 0
@@ -1291,6 +1304,7 @@ def transfer_to(
             
             # Fetch src file and dst directory.
             src_file = self.get_file(path=fp)
+
             if dst_fp in dst_dirs:
                 dst_dir = dst_dirs[dst_fp]
             else:
@@ -1438,7 +1452,7 @@ def _upsert_metadata(self, file_path: str, metadata: dict[str, str]) -> None:
             self.__metadata[abs_path].update({ key: val })
     
 
-    def __new__(cls, *args, **kwargs) -> '_Directory':
+    def __new__(cls, *args, **kwargs):
         '''
         Creates an instance of this class.
 
@@ -2002,7 +2016,7 @@ def load_metadata(self, recursively: bool = False) -> None:
             self.set_metadata(file_path, metadata)
 
 
-class AWSS3Dir(_CloudDir):
+class AmazonS3Dir(_CloudDir):
     '''
     This class represents a virtual directory which resides \
     within an Amazon S3 bucket.
@@ -2111,10 +2125,10 @@ def get_uri(self) -> str:
         return f"s3://{self.get_bucket_name()}/{self.get_path()}"
     
 
-    def get_file(self, path: str) -> AWSS3File:
+    def get_file(self, path: str) -> AmazonS3File:
         '''
         Returns the file residing in the specified \
-        path as a ``AWSS3File`` instance.
+        path as a ``AmazonS3File`` instance.
 
         :param str path: Either the absolute path or the \
             path relative to the directory of the file in \
@@ -2139,16 +2153,16 @@ def get_file(self, path: str) -> AWSS3File:
             raise _IPE(path=path)
         
         path = self._to_absolute(path, replace_sep=False)
-        return AWSS3File._create_file(
+        return AmazonS3File._create_file(
             path=path,
             handler=self._get_handler(),
             metadata=self._get_file_metadata_ref(path))
     
 
-    def get_subdir(self, path: str) -> 'AWSS3Dir':
+    def get_subdir(self, path: str) -> 'AmazonS3Dir':
         '''
         Returns the directory residing in the specified \
-        path as an ``AWSS3Dir`` instance.
+        path as an ``AmazonS3Dir`` instance.
 
         :param str path: Either the absolute path or the \
             path relative to the directory of the subdirectory \
@@ -2180,9 +2194,9 @@ def _create_dir(
         path: str,
         handler: _AWSClientHandler,
         metadata: dict[str, str]
-    ) -> 'AWSS3Dir':
+    ) -> 'AmazonS3Dir':
         '''
-        Creates and returns an ``AWSS3Dir`` instance.
+        Creates and returns an ``AmazonS3Dir`` instance.
 
         :param str path: The path pointing to the directory.
         :param AWSClientHandler handler: An ``AWSClientHandler`` \
@@ -2200,10 +2214,10 @@ class instance.
         return instance
     
 
-    def _get_subdir_impl(self, dir_path: str) -> 'AWSS3Dir':
+    def _get_subdir_impl(self, dir_path: str) -> 'AmazonS3Dir':
         '''
         Returns the directory residing in the specified \
-        path as an ``AWSS3Dir`` instance.
+        path as an ``AmazonS3Dir`` instance.
 
         :param str dir_path: Either the absolute path \
             or the path relative to the directory of the \
@@ -2220,7 +2234,7 @@ def _get_subdir_impl(self, dir_path: str) -> 'AWSS3Dir':
             metadata=self._get_metadata_ref())
     
 
-    def __enter__(self) -> 'AWSS3Dir':
+    def __enter__(self) -> 'AmazonS3Dir':
         '''
         Enter the runtime context related to this instance.
         '''
@@ -2368,7 +2382,6 @@ def get_file(self, path: str) -> AzureBlobFile:
                 * Wrong: ``/path/to/file.txt``
                 * Right: ``path/to/file.txt``
         '''
-
         if not self.path_exists(path):
             raise _IPE(path=path)
         if not self.is_file(path):
diff --git a/pyproject.toml b/pyproject.toml
index 5f0099c..fd3e74f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "fluke-api"
-version = "0.3.0"
+version = "0.4.0"
 authors = [
   {email = "manosstoumpos@gmail.com"},
   {name = "Manos Stoumpos"}
@@ -16,17 +16,18 @@ classifiers = [
     "Operating System :: OS Independent"
 ]
 dependencies = [
-  "azure-identity == 1.12.0",
-  "azure-storage-blob == 12.15.0",
-  "boto3 == 1.26.84",
-  "paramiko == 3.0.0",
+  "azure-identity == 1.13.0",
+  "azure-storage-blob == 12.17.0",
+  "azure-storage-queue == 12.6.0",
+  "boto3 == 1.28.0",
+  "paramiko == 3.2.0",
   "tqdm == 4.65.0"
 ]
 
 [project.optional-dependencies]
 tests = [
   "coverage == 7.2.1",
-  "moto[s3] == 4.1.4"
+  "moto[s3,sqs] == 4.1.4"
 ]
 
 [project.urls]


Test patch:
diff --git a/tests/test_auth.py b/tests/test_auth.py
index 766d330..16808aa 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -95,8 +95,8 @@ class TestAzureAuth(unittest.TestCase):
     CLIENT_ID = "CLIENT_ID"
     CLIENT_SECRET = "CLIENT_SECRET"
     CONNECTION_STRING = "CONNECTION_STRING"
-
-    def test_get_credentials(self):
+        
+    def test_get_credentials_from_service_principal(self):
         credentials = {
             'account_url': self.ACCOUNT_URL,
             'tenant_id': self.TENANT_ID,
@@ -104,10 +104,10 @@ def test_get_credentials(self):
             'client_secret': self.CLIENT_SECRET
         }
         self.assertEqual(
-            AzureAuth(**credentials).get_credentials(),
+            AzureAuth.from_service_principal(**credentials).get_credentials(),
             credentials)
         
-    def test_from_connection_string(self):
+    def test_get_credentials_from_connection_string(self):
         credentials = { 'conn_string': self.CONNECTION_STRING }
         self.assertEqual(
             AzureAuth.from_conn_string(**credentials).get_credentials(),
diff --git a/tests/test_queues.py b/tests/test_queues.py
new file mode 100644
index 0000000..7bc16b1
--- /dev/null
+++ b/tests/test_queues.py
@@ -0,0 +1,465 @@
+import os
+import io
+import sys
+import time
+import shutil
+import unittest
+from typing import Optional, Iterator, Callable
+from unittest.mock import Mock, patch
+
+
+import boto3
+from moto import mock_sqs
+
+
+from fluke.auth import AWSAuth, AzureAuth
+from fluke.queues import AmazonSQSQueue, AzureStorageQueue
+
+
+QUEUE = "test-queue"
+STORAGE_ACCOUNT = "account"
+STORAGE_ACCOUNT_URL = f"https://{STORAGE_ACCOUNT}.blob.core.windows.net/"
+STORAGE_ACCOUNT_CONN_STRING = f"DefaultEndpointsProtocol=https;AccountName={STORAGE_ACCOUNT};..."
+
+
+def get_azure_auth_instance(from_conn_string: bool) -> AzureAuth:
+    '''
+    Returns a dummy ``AzureAuth`` instance.
+
+    :param bool from_conn_string: Indicates whether \
+        to construct the instance using a connection \
+        string, or the regular Azure credentials.
+    '''
+    return (
+        AzureAuth.from_conn_string(conn_string=STORAGE_ACCOUNT_CONN_STRING)
+        if from_conn_string else 
+        AzureAuth.from_service_principal(
+            account_url=STORAGE_ACCOUNT_URL,
+            tenant_id='',
+            client_id='',
+            client_secret='')
+    )
+
+
+class MockQueueClient():
+
+    def simulate_latency(func: Callable):
+        '''
+        This function is to be used as a decorator \
+        function in order to simulate network latency \
+        in mocked classes, so that cache-related methods \
+        can be effectively tested.
+        '''
+        def wrapper(*args, **kwargs):
+            time.sleep(0.02)
+            return func(*args, **kwargs)
+        return wrapper
+    
+    class MockQueueProperties():
+
+        def __init__(self, name: str) -> None:
+            self.name = name
+            self.approximate_message_count = 0
+
+    class MockItemPages(list):
+
+        class MockQueueMessage():
+
+            def __init__(self, id: str, content: str) -> None:
+                self.id = id
+                self.content = content
+                self.pop_receipt = None
+
+        def __init__(
+            self,
+            messages: list[MockQueueMessage],
+            messages_per_page: int,
+            max_messages: int
+        ) -> None:
+            self.__messages = messages
+            self.__messages_per_page = messages_per_page
+            self.__max_messages = len(messages) if max_messages is None else max_messages
+
+        def by_page(self) -> Iterator[MockQueueMessage]:
+            k = 0
+            for i in range(0, self.__max_messages, self.__messages_per_page):
+                # This means that messages are being polled, therefore deleted.
+                if len(self.__messages[i*self.__messages_per_page:(i+1)*self.__messages_per_page]) == 0:
+                    k += self.__messages_per_page
+                idx = i-k
+                for j in range(i, len(self.__messages[idx*self.__messages_per_page:(idx+1)*self.__messages_per_page])):
+                    self.__messages[j].pop_receipt = str(j)
+                yield (msg for msg in self.__messages[idx*self.__messages_per_page:(idx+1)*self.__messages_per_page])
+                
+
+    def __init__(self, queue_name: str):
+        self.queue_name = queue_name
+        self.properties = __class__.MockQueueProperties(queue_name)
+        self.messages: list[__class__.MockItemPages.MockQueueMessage] = []
+
+
+    @staticmethod
+    def get_mock_methods() -> dict[str, Mock]:
+        '''
+        Returns a dictionary containing all the \
+        methods that are to be mocked.
+        '''
+        return {
+            'azure.identity.ClientSecretCredential.__init__': Mock(return_value=None),
+            'azure.storage.queue.QueueClient.__init__': Mock(return_value=None),
+            'azure.storage.queue.QueueClient.from_connection_string': Mock(
+                return_value=TestAzureStorageQueue.queue
+            )
+        }
+
+    @simulate_latency
+    def get_queue_properties(self) -> MockQueueProperties:
+        return self.properties
+    
+    @simulate_latency
+    def send_message(self, content: str) -> None:
+        self.messages.append(__class__.MockItemPages.MockQueueMessage(
+            id=str(len(self.messages)),
+            content=content))
+        self.properties.approximate_message_count += 1
+
+    @simulate_latency
+    def peek_messages(self, max_messages: int) -> None:
+        return self.messages[:max_messages]
+    
+    @simulate_latency
+    def receive_messages(
+        self,
+        messages_per_page: int,
+        max_messages: int,
+        visibility_timeout: int
+    ) -> MockItemPages[MockItemPages.MockQueueMessage]:
+        return __class__.MockItemPages(
+            self.messages,
+            messages_per_page,
+            max_messages
+        )
+    
+    @simulate_latency
+    def delete_message(self, id: str, pop_receipt: str) -> None:
+        delete_idx = None
+        for i, msg in enumerate(self.messages):
+            if msg.id == id and msg.pop_receipt == pop_receipt:
+                delete_idx = i
+        
+        if delete_idx is None:
+            raise Exception("Message not found")
+        
+        self.messages.pop(delete_idx)
+        self.properties.approximate_message_count -= 1
+
+    @simulate_latency
+    def clear_messages(self) -> None:
+        self.messages = list()
+        self.properties.approximate_message_count = 0
+
+    @simulate_latency
+    def close(self):
+        pass
+
+
+class TestAmazonSQSQueue(unittest.TestCase):
+
+    MOCK_SQS = mock_sqs()
+    QUEUE = None
+
+    @staticmethod
+    def build_queue() -> AmazonSQSQueue:
+        return AmazonSQSQueue(**{
+            'auth': AWSAuth(
+                aws_access_key_id='',
+                aws_secret_access_key='',
+                region=AWSAuth.Region.EUROPE_WEST_1),
+            'queue': QUEUE
+        })
+    
+    def get_num_messages(self) -> int:
+        self.QUEUE.load()
+        return (
+            int(self.QUEUE.attributes['ApproximateNumberOfMessages']) +
+            int(self.QUEUE.attributes['ApproximateNumberOfMessagesNotVisible'])
+        )
+
+    def fetch_messages(self) -> list[str]:
+        messages = []
+        while len(batch := self.QUEUE.receive_messages(
+            AttributeNames=['All'],
+            MaxNumberOfMessages=10,
+            VisibilityTimeout=60,
+            WaitTimeSeconds=0)
+        ) > 0:
+            resp = self.QUEUE.delete_messages(Entries=[
+                {
+                    'Id': str(i),
+                    'ReceiptHandle': msg.receipt_handle
+                } for i, msg in enumerate(batch)])
+            
+            for i in map(lambda d: int(d['Id']), resp['Successful']):
+                messages.append(batch[i].body)
+            
+        return messages
+    
+    def send_message(self, message: str) -> None:
+        self.QUEUE.send_message(
+            MessageBody=message,
+            DelaySeconds=0)
+
+    def send_messages(self, messages: list[str]) -> None:
+        for msg in messages:
+            self.send_message(msg)
+
+    def purge(self) -> None:
+        self.QUEUE.purge()
+
+    def setUp(self):
+        self.MOCK_SQS.start()
+        self.QUEUE = boto3.resource(
+            'sqs',
+            endpoint_url="https://sqs.eu-west-1.amazonaws.com",
+            region_name="eu-west-1"
+        ).create_queue(QueueName=QUEUE)
+
+    def tearDown(self):
+        self.MOCK_SQS.stop()
+
+    def test_is_open_on_True(self):
+        with self.build_queue() as queue:
+            self.assertEqual(queue.is_open(), True)
+
+    def test_is_open_on_False(self):
+        queue = self.build_queue()
+        queue.close()
+        self.assertEqual(queue.is_open(), False)
+
+    def test_get_name(self):
+        with self.build_queue() as queue:
+            self.assertEqual(queue.get_name(), QUEUE)
+
+    def test_count(self):
+        num_messages = 5
+        self.send_messages([str(i) for i in range(num_messages)])
+        with self.build_queue() as queue:
+            self.assertEqual(queue.count(), num_messages)
+        self.purge()
+
+    def test_push(self):
+        message = "Hello"
+        with self.build_queue() as queue:
+            self.assertTrue(queue.push(message))
+            self.assertEqual(self.fetch_messages()[0], message)
+
+    def test_peek(self):
+        messages = set(str(i) for i in range(5))
+        self.send_messages(messages)
+        with self.build_queue() as queue:
+            self.assertTrue(set(queue.peek()).issubset(messages))
+        self.purge()
+
+    def test_poll(self):
+        messages = set(str(i) for i in range(5))
+        self.send_messages(messages)
+        with self.build_queue() as queue:
+            fetched = set(msg for batch in queue.poll() for msg in batch)
+            self.assertSetEqual(fetched, messages)
+            self.assertEqual(self.get_num_messages(), 0)
+        self.purge()
+
+    def test_poll_on_num_messages(self):
+        messages = set(str(i) for i in range(5))
+        num_messages = 3
+        self.send_messages(messages)
+        with self.build_queue() as queue:
+            fetched = set(msg for batch in queue.poll(num_messages=num_messages) for msg in batch)
+            self.assertTrue(fetched.issubset(messages))
+            self.assertEqual(self.get_num_messages(), len(messages) - num_messages)
+        self.purge()
+
+    def test_poll_on_batch_size(self):
+        messages = set(str(i) for i in range(29))
+        batch_size = 10
+        batch_sizes = [10, 10, 9]
+        self.send_messages(messages)
+        counter = 0
+        with self.build_queue() as queue:
+            for batch in queue.poll(batch_size=batch_size):
+                self.assertEqual(len(batch), batch_sizes[counter])
+                counter += 1
+        self.purge()
+
+    def test_poll_on_pre_delivery_delete_set_to_False(self):
+        num_messages = 5
+        messages = set(str(i) for i in range(num_messages))
+        self.send_messages(messages)
+        with self.build_queue() as queue:
+            for _ in queue.poll(pre_delivery_delete=False):
+                # Assert that messages have not been deleted yet.
+                self.assertEqual(self.get_num_messages(), num_messages)
+        self.purge()
+
+    def test_poll_on_pre_delivery_delete_set_to_True(self):
+        num_messages = 5
+        messages = set(str(i) for i in range(num_messages))
+        self.send_messages(messages)
+        with self.build_queue() as queue:
+            for _ in queue.poll(pre_delivery_delete=True):
+                # Assert that messages have already been deleted.
+                self.assertEqual(self.get_num_messages(), 0)
+        self.purge()
+
+    def test_clear(self):
+        self.send_messages(set(str(i) for i in range(50)))
+        with self.build_queue() as queue:
+            queue.clear()
+            self.assertEqual(self.get_num_messages(), 0)
+        self.purge()
+
+
+class TestAzureStorageQueue(unittest.TestCase):
+
+    queue = MockQueueClient(queue_name=QUEUE)
+
+    def get_num_messages(self) -> int:
+        return len(self.queue.messages)
+    
+    def fetch_messages(self) -> list[str]:
+        messages = []
+        while len(batch := self.queue.receive_messages(
+            messages_per_page=10,
+            max_messages=None,
+            visibility_timeout=60)
+        ) > 0:
+            for msg in batch:
+                try:
+                    self.queue.delete_message(msg.id, msg.pop_recipt)
+                    messages.append(msg.content)
+                except:
+                    continue
+            
+        return messages
+    
+    def send_message(self, message: str) -> None:
+        self.queue.send_message(message)
+
+    def send_messages(self, messages: list[str]) -> None:
+        for msg in messages:
+            self.queue.send_message(msg)
+
+    def purge(self) -> None:
+        self.queue.clear_messages()
+
+    def setUp(self):
+        for k, v in MockQueueClient.get_mock_methods().items():
+            patch(k, v).start()
+
+    def tearDown(self):
+        patch.stopall()
+
+    @staticmethod
+    def build_queue(
+        from_conn_string: str = STORAGE_ACCOUNT_CONN_STRING
+    ) -> AzureStorageQueue:
+        return AzureStorageQueue(**{
+            'auth': get_azure_auth_instance(from_conn_string),
+            'queue': QUEUE
+        })
+
+    def test_is_open_on_True(self):
+        with self.build_queue() as queue:
+            self.assertEqual(queue.is_open(), True)
+
+    def test_is_open_on_False(self):
+        queue = self.build_queue()
+        queue.close()
+        self.assertEqual(queue.is_open(), False)
+
+    def test_get_name(self):
+        with self.build_queue() as queue:
+            self.assertEqual(queue.get_name(), QUEUE)
+
+    def test_count(self):
+        num_messages = 5
+        self.send_messages([str(i) for i in range(num_messages)])
+        with self.build_queue() as queue:
+            self.assertEqual(queue.count(), num_messages)
+        self.purge()
+
+    def test_push(self):
+        message = "Hello"
+        with self.build_queue() as queue:
+            self.assertTrue(queue.push(message))
+            self.assertEqual(queue.peek()[0], message)
+        self.purge()
+
+    def test_peek(self):
+        messages = set(str(i) for i in range(5))
+        self.send_messages(messages)
+        with self.build_queue() as queue:
+            self.assertTrue(set(queue.peek()).issubset(messages))
+        self.purge()
+
+    def test_poll(self):
+        messages = set(str(i) for i in range(5))
+        self.send_messages(messages)
+        with self.build_queue() as queue:
+            fetched = set(msg for batch in queue.poll() for msg in batch)
+            self.assertSetEqual(fetched, messages)
+            self.assertEqual(queue.count(), 0)
+        self.purge()
+
+    def test_poll_on_num_messages(self):
+        messages = set(str(i) for i in range(5))
+        num_messages = 3
+        self.send_messages(messages)
+        with self.build_queue() as queue:
+            fetched = set(msg for batch in queue.poll(num_messages=num_messages) for msg in batch)
+            self.assertTrue(fetched.issubset(messages))
+            self.assertEqual(queue.count(), len(messages) - num_messages)
+        self.purge()
+
+    def test_poll_on_batch_size(self):
+        messages = set(str(i) for i in range(29))
+        batch_size = 10
+        batch_sizes = [10, 10, 9]
+        self.send_messages(messages)
+        counter = 0
+        with self.build_queue() as queue:
+            for batch in queue.poll(batch_size=batch_size):
+                self.assertEqual(len(batch), batch_sizes[counter])
+                counter += 1
+        self.purge()
+
+    def test_poll_on_pre_delivery_delete_set_to_False(self):
+        num_messages = 5
+        messages = set(str(i) for i in range(num_messages))
+        self.send_messages(messages)
+        with self.build_queue() as queue:
+            for _ in queue.poll(pre_delivery_delete=False):
+                # Assert that messages have not been deleted yet.
+                self.assertEqual(queue.count(), num_messages)
+        self.purge()
+
+    def test_poll_on_pre_delivery_delete_set_to_True(self):
+        num_messages = 5
+        messages = set(str(i) for i in range(num_messages))
+        self.send_messages(messages)
+        with self.build_queue() as queue:
+            for _ in queue.poll(pre_delivery_delete=True):
+                # Assert that messages have already been deleted.
+                self.assertEqual(queue.count(), 0)
+        self.purge()
+
+    def test_clear(self):
+        self.send_messages(set(str(i) for i in range(50)))
+        with self.build_queue() as queue:
+            queue.clear()
+            self.assertEqual(queue.count(), 0)
+        self.purge()
+
+
+if __name__=="__main__":
+    unittest.main()
\ No newline at end of file
diff --git a/tests/test_storage.py b/tests/test_storage.py
index 3e11c37..8039224 100644
--- a/tests/test_storage.py
+++ b/tests/test_storage.py
@@ -15,7 +15,7 @@
 
 from fluke.auth import RemoteAuth, AWSAuth, AzureAuth
 from fluke.storage import LocalFile, LocalDir, \
-    RemoteFile, RemoteDir, AWSS3File, AWSS3Dir, \
+    RemoteFile, RemoteDir, AmazonS3File, AmazonS3Dir, \
     AzureBlobFile, AzureBlobDir
 from fluke._exceptions import OverwriteError
 from fluke._exceptions import InvalidPathError
@@ -76,7 +76,7 @@ def get_azure_auth_instance(from_conn_string: bool) -> AzureAuth:
     return (
         AzureAuth.from_conn_string(conn_string=STORAGE_ACCOUNT_CONN_STRING)
         if from_conn_string else 
-        AzureAuth(
+        AzureAuth.from_service_principal(
             account_url=STORAGE_ACCOUNT_URL,
             tenant_id='',
             client_id='',
@@ -820,12 +820,12 @@ def test_transfer_to_on_include_metadata_set_to_false(self):
            # Copy file into dir, including its metadata.
             file.set_metadata(metadata=METADATA)
             dir_path = REL_DIR_PATH
-            with (AWSS3Dir(
+            with (AmazonS3Dir(
                     auth=get_aws_auth_instance(),
                     bucket=BUCKET,
-                    path=dir_path) as aws_dir
+                    path=dir_path) as s3_dir
             ):
-                file.transfer_to(dst=aws_dir, include_metadata=False)
+                file.transfer_to(dst=s3_dir, include_metadata=False)
             # Gain access to the uploaded object.
             file_path = join_paths(dir_path, FILE_NAME)
             obj = get_aws_s3_object(BUCKET, file_path)
@@ -848,12 +848,12 @@ def test_transfer_to_on_include_metadata_set_to_true(self):
             new_metadata = {'2': '2'}
             file.set_metadata(metadata=new_metadata)
             dir_path = REL_DIR_PATH
-            with (AWSS3Dir(
+            with (AmazonS3Dir(
                     auth=get_aws_auth_instance(),
                     bucket=BUCKET,
-                    path=dir_path) as aws_dir
+                    path=dir_path) as s3_dir
             ):
-                file.transfer_to(dst=aws_dir, include_metadata=True)
+                file.transfer_to(dst=s3_dir, include_metadata=True)
             # Gain access to the uploaded object.
             file_path = join_paths(dir_path, FILE_NAME)
             obj = get_aws_s3_object(BUCKET, file_path)
@@ -924,7 +924,7 @@ def test_get_size_from_cache_on_time(self):
             self.assertGreater(normal_time, cache_time)
 
 
-class TestAWSS3File(unittest.TestCase):
+class TestAmazonS3File(unittest.TestCase):
 
     MOCK_S3 = mock_s3()
 
@@ -959,8 +959,8 @@ def tearDown(self):
         patch.stopall()
     
     @staticmethod
-    def build_file(path: str = REL_FILE_PATH, cache: bool = False) -> AWSS3File:
-        return AWSS3File(**{
+    def build_file(path: str = REL_FILE_PATH, cache: bool = False) -> AmazonS3File:
+        return AmazonS3File(**{
             'auth': get_aws_auth_instance(),
             'bucket': BUCKET,
             'path': path,
@@ -1128,12 +1128,12 @@ def test_transfer_to_on_include_metadata_set_to_false(self):
             dir_path = REL_DIR_PATH
             file_path = join_paths(dir_path, FILE_NAME)
             # Copy file into dir, including its metadata.
-            with (AWSS3Dir(
+            with (AmazonS3Dir(
                     auth=get_aws_auth_instance(),
                     bucket=BUCKET,
                     path=dir_path)
-            as aws_dir):
-                file.transfer_to(dst=aws_dir, include_metadata=False)
+            as s3_dir):
+                file.transfer_to(dst=s3_dir, include_metadata=False)
             # Gain access to the uploaded object.
             obj = get_aws_s3_object(BUCKET, file_path)
             # Assert that metadata were copied to the object.
@@ -1146,14 +1146,14 @@ def test_transfer_to_on_include_metadata_set_to_true(self):
             dir_path = REL_DIR_PATH
             file_path = join_paths(dir_path, FILE_NAME)
             # Copy file into dir, including its metadata.
-            with (AWSS3Dir(
+            with (AmazonS3Dir(
                     auth=get_aws_auth_instance(),
                     bucket=BUCKET,
                     path=dir_path)
-            as aws_dir):
+            as s3_dir):
                 new_metadata = {'2': '2'}
                 file.set_metadata(new_metadata)
-                file.transfer_to(dst=aws_dir, include_metadata=True)
+                file.transfer_to(dst=s3_dir, include_metadata=True)
             # Gain access to the uploaded object.
             obj = get_aws_s3_object(BUCKET, file_path)
             # Assert that metadata were copied to the object.
@@ -1440,12 +1440,12 @@ def test_transfer_to_on_include_metadata_set_to_false(self):
            # Copy file into dir, including its metadata.
             file.set_metadata(metadata=METADATA)
             dir_path = REL_DIR_PATH
-            with (AWSS3Dir(
+            with (AmazonS3Dir(
                     auth=get_aws_auth_instance(),
                     bucket=BUCKET,
-                    path=dir_path) as aws_dir
+                    path=dir_path) as s3_dir
             ):
-                file.transfer_to(dst=aws_dir, include_metadata=False)
+                file.transfer_to(dst=s3_dir, include_metadata=False)
             # Gain access to the uploaded object.
             file_path = join_paths(dir_path, FILE_NAME)
             obj = get_aws_s3_object(BUCKET, file_path)
@@ -1468,12 +1468,12 @@ def test_transfer_to_on_include_metadata_set_to_true(self):
             new_metadata = {'2': '2'}
             file.set_metadata(metadata=new_metadata)
             dir_path = REL_DIR_PATH
-            with (AWSS3Dir(
+            with (AmazonS3Dir(
                     auth=get_aws_auth_instance(),
                     bucket=BUCKET,
-                    path=dir_path) as aws_dir
+                    path=dir_path) as s3_dir
             ):
-                file.transfer_to(dst=aws_dir, include_metadata=True)
+                file.transfer_to(dst=s3_dir, include_metadata=True)
             # Gain access to the uploaded object.
             file_path = join_paths(dir_path, FILE_NAME)
             obj = get_aws_s3_object(BUCKET, file_path)
@@ -1847,6 +1847,31 @@ def test_transfer_to_on_chunk_size(self):
         # Remove temporary directory.
         shutil.rmtree(tmp_dir_path)
 
+    def test_transfer_to_on_filter(self):
+        # Create a temporary dictionary.
+        tmp_dir_path = to_abs(REL_DIR_PATH.replace('dir', TMP_DIR_NAME))
+        os.mkdir(tmp_dir_path)
+        # Copy the directory's contents into this tmp directory.
+        self.build_dir(path=ABS_DIR_PATH).transfer_to(
+            dst=self.build_dir(path=tmp_dir_path),
+            filter=lambda x: not x.endswith('.txt'))
+        # Assert that the two directories contain the same files.
+        original = [s for s in sorted(os.listdir(ABS_DIR_PATH))
+                    if os.path.isfile(s) and not s.endswith('.txt')]
+        copies = [s for s in sorted(os.listdir(tmp_dir_path))]
+        # 1. Assert number of copied files are the same.
+        self.assertEqual(len(original), len(copies))
+        # 2. Iterate over all files.
+        for ofp, cfp in zip(original, copies):
+            # Assert their contents are the same.
+            with (
+                open(file=join_paths(ABS_DIR_PATH, ofp), mode='rb') as of,
+                open(file=join_paths(tmp_dir_path, cfp), mode='rb') as cp
+            ):
+                self.assertEqual(of.read(), cp.read())
+        # Remove temporary directory.
+        shutil.rmtree(tmp_dir_path)
+
     def test_transfer_to_on_overwrite_set_to_false(self):
         # Create a copy of the directory.
         tmp_dir_name = TMP_DIR_NAME
@@ -1929,13 +1954,13 @@ def test_transfer_to_on_aws_cloud_dir_include_metadata_set_to_true(self):
             # Copy directory contents including metadata
             filename, metadata = DIR_FILE_NAME, {'2': '2'}
             dir = self.build_dir(path=ABS_DIR_PATH)
-            with (TestAWSS3Dir.build_dir() as aws_dir):
+            with (TestAmazonS3Dir.build_dir() as s3_dir):
                 dir.set_metadata(file_path=filename, metadata=metadata)
-                dir.transfer_to(dst=aws_dir, overwrite=True, include_metadata=True)
+                dir.transfer_to(dst=s3_dir, overwrite=True, include_metadata=True)
                 # Assert that the object's metadata have been modified.
                 self.assertEqual(
                     get_aws_s3_object(
-                        aws_dir.get_bucket_name(),
+                        s3_dir.get_bucket_name(),
                         REL_DIR_FILE_PATH).metadata,
                     metadata)
         
@@ -2297,6 +2322,32 @@ def test_transfer_to_on_chunk_size(self):
         # Remove temporary directory.
         shutil.rmtree(tmp_dir_path)
 
+    def test_transfer_to_on_filter(self):
+        # Create a temporary dictionary.
+        tmp_dir_path = to_abs(REL_DIR_PATH.replace('dir', TMP_DIR_NAME))
+        os.mkdir(tmp_dir_path)
+        # Copy the directory's contents into this tmp directory.
+        with self.build_dir() as dir:
+            dir.transfer_to(
+                dst=LocalDir(path=tmp_dir_path),
+                filter=lambda x: not x.endswith('.txt'))
+        # Assert that the two directories contain the same files.
+        original = [s for s in sorted(os.listdir(ABS_DIR_PATH))
+                    if os.path.isfile(s) and not s.endswith('.txt')]
+        copies = [s for s in sorted(os.listdir(tmp_dir_path))]
+        # 1. Assert number of copied files are the same.
+        self.assertEqual(len(original), len(copies))
+        # 2. Iterate over all files.
+        for ofp, cfp in zip(original, copies):
+            # Assert their contents are the same.
+            with (
+                open(file=join_paths(ABS_DIR_PATH, ofp), mode='rb') as of,
+                open(file=join_paths(tmp_dir_path, cfp), mode='rb') as cp
+            ):
+                self.assertEqual(of.read(), cp.read())
+        # Remove temporary directory.
+        shutil.rmtree(tmp_dir_path)
+
     def test_transfer_to_on_overwrite_set_to_false(self):
         # Create a copy of the directory.
         tmp_dir_name = TMP_DIR_NAME
@@ -2693,7 +2744,7 @@ def test_subdir_shared_cache_on_cache_via_subdir(self):
             self.assertGreater(normal_time, cache_time)
 
 
-class TestAWSS3Dir(unittest.TestCase):
+class TestAmazonS3Dir(unittest.TestCase):
 
     MOCK_S3 = mock_s3()
 
@@ -2749,8 +2800,8 @@ def build_dir(
         path: str = REL_DIR_PATH,
         cache: bool = False,
         create_if_missing: bool = False
-    ) -> AWSS3Dir:
-        return AWSS3Dir(**{
+    ) -> AmazonS3Dir:
+        return AmazonS3Dir(**{
             'auth': get_aws_auth_instance(),
             'bucket': BUCKET,
             'path': path,
@@ -3075,6 +3126,33 @@ def test_transfer_to_on_chunk_size(self):
         # Remove temporary directory.
         shutil.rmtree(tmp_dir_path)
 
+    def test_transfer_to_on_filter(self):
+        # Create a temporary dictionary.
+        tmp_dir_path = to_abs(REL_DIR_PATH.replace('dir', TMP_DIR_NAME))
+        os.mkdir(tmp_dir_path)
+        # Copy the directory's contents into this tmp directory.
+        with self.build_dir() as dir:
+            dir.transfer_to(
+                dst=LocalDir(path=tmp_dir_path),
+                filter=lambda x: not x.endswith('.txt'))
+        # Assert that the two directories contain the same files.
+        original = [s for s in sorted(os.listdir(ABS_DIR_PATH))
+                    if os.path.isfile(s) and not s.endswith('.txt')]
+        copies = [s for s in sorted(os.listdir(tmp_dir_path))]
+        # 1. Assert number of copied files are the same.
+        self.assertEqual(len(original), len(copies))
+        # 2. Iterate over all files.
+        for ofp, cfp in zip(original, copies):
+            # Assert their contents are the same.
+            with (
+                io.BytesIO() as buffer,
+                open(file=join_paths(tmp_dir_path, cfp), mode='rb') as cp
+            ):
+                get_aws_s3_object(BUCKET, ofp).download_fileobj(buffer)
+                self.assertEqual(buffer.getvalue(), cp.read())
+        # Remove temporary directory.
+        shutil.rmtree(tmp_dir_path)
+
     def test_transfer_to_on_overwrite_set_to_false(self):
         # Create a copy of the directory.
         tmp_dir_name = TMP_DIR_NAME
@@ -3178,9 +3256,9 @@ def test_transfer_to_as_dst(self):
         src_file = TestLocalFile.build_file()
         # Get tmp dir path (already created in bucket).
         tmp_dir_path = REL_DIR_PATH.replace('dir', TMP_DIR_NAME)
-        with self.build_dir(path=tmp_dir_path) as aws_dir:
+        with self.build_dir(path=tmp_dir_path) as s3_dir:
             # Copy file into dir.
-            src_file.transfer_to(dst=aws_dir)
+            src_file.transfer_to(dst=s3_dir)
             # Confirm that file was indeed copied.
             copy_path = join_paths(tmp_dir_path, FILE_NAME)
             with (
@@ -3199,9 +3277,9 @@ def test_transfer_to_as_dst_on_chunk_size(self):
         src_file = TestLocalFile.build_file()
         # Get tmp dir path (already created in bucket).
         tmp_dir_path = REL_DIR_PATH.replace('dir', TMP_DIR_NAME)
-        with self.build_dir(path=tmp_dir_path) as aws_dir:
+        with self.build_dir(path=tmp_dir_path) as s3_dir:
             # Copy file into dir.
-            src_file.transfer_to(dst=aws_dir, chunk_size=1000)
+            src_file.transfer_to(dst=s3_dir, chunk_size=1000)
             # Confirm that file was indeed copied.
             copy_path = join_paths(tmp_dir_path, FILE_NAME)
             with (
@@ -3222,9 +3300,9 @@ def test_transfer_to_as_dst_on_include_metadata(self):
         src_file.set_metadata(metadata=metadata)
         # Get tmp dir path (already created in bucket).
         tmp_dir_path = REL_DIR_PATH.replace('dir', TMP_DIR_NAME)
-        with self.build_dir(path=tmp_dir_path) as aws_dir:
+        with self.build_dir(path=tmp_dir_path) as s3_dir:
             # Copy file into dir.
-            src_file.transfer_to(dst=aws_dir, include_metadata=True)
+            src_file.transfer_to(dst=s3_dir, include_metadata=True)
             # Confirm that metadata was indeed assigned.
             copy_path = join_paths(tmp_dir_path, FILE_NAME)
             self.assertEqual(
@@ -3238,10 +3316,10 @@ def test_transfer_to_as_dst_on_chunk_size_and_include_metadata(self):
         src_file.set_metadata(metadata=metadata)
         # Get tmp dir path (already created in bucket).
         tmp_dir_path = REL_DIR_PATH.replace('dir', TMP_DIR_NAME)
-        with self.build_dir(path=tmp_dir_path) as aws_dir:
+        with self.build_dir(path=tmp_dir_path) as s3_dir:
             # Copy file into dir.
             src_file.transfer_to(
-                dst=aws_dir,
+                dst=s3_dir,
                 include_metadata=True,
                 chunk_size=1000)
             # Confirm that metadata was indeed assigned.
@@ -3903,6 +3981,33 @@ def test_transfer_to_on_chunk_size(self):
         # Remove temporary directory.
         shutil.rmtree(tmp_dir_path)
 
+
+    def test_transfer_to_on_filter(self):
+        # Create a temporary dictionary.
+        tmp_dir_path = to_abs(REL_DIR_PATH.replace('dir', TMP_DIR_NAME))
+        os.mkdir(tmp_dir_path)
+        # Copy the directory's contents into this tmp directory.
+        with self.build_dir() as dir:
+            dir.transfer_to(
+                dst=LocalDir(path=tmp_dir_path),
+                filter=lambda x: not x.endswith('.txt'))
+        # Assert that the two directories contain the same files.
+        original = [s for s in sorted(os.listdir(ABS_DIR_PATH))
+                    if os.path.isfile(s) and not s.endswith('.txt')]
+        copies = [s for s in sorted(os.listdir(tmp_dir_path))]
+        # 1. Assert number of copied files are the same.
+        self.assertEqual(len(original), len(copies))
+        # 2. Iterate over all files.
+        for ofp, cfp in zip(original, copies):
+            # Assert their contents are the same.
+            with (
+                open(file=join_paths(ABS_DIR_PATH, ofp), mode='rb') as of,
+                open(file=join_paths(tmp_dir_path, cfp), mode='rb') as cp
+            ):
+                self.assertEqual(of.read(), cp.read())
+        # Remove temporary directory.
+        shutil.rmtree(tmp_dir_path)
+
     def test_transfer_to_on_overwrite_set_to_false(self):
         # Create a copy of the directory.
         tmp_dir_name = TMP_DIR_NAME