diff --git a/README.md b/README.md
index 63082cb08..64307869d 100644
--- a/README.md
+++ b/README.md
@@ -57,6 +57,50 @@ Please note that on Mac, once a private key is used with a certificate, that cer
 static: certificate has an existing certificate-key pair that was previously imported into the Keychain. Using key from Keychain instead of the one provided.
 ```
 
+## Logging
+
+aws-crt-python provides two ways to configure logging from the CRT's native C libraries.
+
+### Option 1: Python logging integration (recommended)
+
+Route CRT log messages through Python's standard `logging` module. Each CRT subsystem gets a hierarchical logger under `awscrt` (e.g. `awscrt.io.event-loop`, `awscrt.http.http-connection`, `awscrt.s3.S3Client`), so you can filter by package or subsystem.
+
+```python
+import logging
+from awscrt.logging import init_logging, CRT_LOG_FORMAT
+
+handler = logging.StreamHandler()
+handler.setFormatter(logging.Formatter(CRT_LOG_FORMAT))
+logging.getLogger('awscrt').addHandler(handler)
+init_logging(logging.DEBUG)
+```
+
+To filter to a specific CRT package:
+
+```python
+# Only show IO logs at DEBUG, silence everything else
+logging.getLogger('awscrt').setLevel(logging.WARNING)
+logging.getLogger('awscrt.io').setLevel(logging.DEBUG)
+```
+
+### Option 2: Direct file/stdout logging
+
+Write CRT log output directly to a file, stdout, or stderr. This bypasses Python's logging module and is useful for quick debugging.
+
+```python
+from awscrt.io import LogLevel, init_logging
+
+init_logging(LogLevel.Debug, 'stdout')   # or 'stderr', or a file path
+```
+
+### Which should I use?
+
+Use `awscrt.logging.init_logging` (Option 1) if you want CRT logs integrated with your application's existing Python logging setup, with per-subsystem filtering.
+
+Use `awscrt.io.init_logging` (Option 2) if you just want to dump all CRT logs to a file or console quickly.
+
+These are mutually exclusive — use one or the other, not both.
+
 ## Crash Handler
 
 You can enable the crash handler by setting the environment variable `AWS_CRT_CRASH_HANDLER=1` . This will print the callstack to `stderr` in the event of a fatal error.
diff --git a/awscrt/__init__.py b/awscrt/__init__.py
index f340fe98f..dbe9eb2c8 100644
--- a/awscrt/__init__.py
+++ b/awscrt/__init__.py
@@ -9,6 +9,7 @@
     'crypto',
     'http',
     'io',
+    'logging',
     'mqtt',
     'mqtt5',
     'mqtt_request_response',
diff --git a/awscrt/logging.py b/awscrt/logging.py
new file mode 100644
index 000000000..e7f210b7f
--- /dev/null
+++ b/awscrt/logging.py
@@ -0,0 +1,205 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0.
+
+"""Python logging integration for the AWS CRT."""
+
+import logging
+import threading
+from enum import IntEnum
+
+import _awscrt
+from awscrt.io import LogLevel
+
+__all__ = ['init_logging', 'set_log_level', 'logf', 'LogSubject', 'CRT_LOG_FORMAT']
+
+_CRT_TO_PY_LEVEL = {
+    0: logging.NOTSET,
+    1: logging.CRITICAL,
+    2: logging.ERROR,
+    3: logging.WARNING,
+    4: logging.INFO,
+    5: logging.DEBUG,
+    6: logging.DEBUG,
+}
+
+_PY_TO_CRT_LEVEL = {
+    logging.NOTSET: LogLevel.NoLogs,
+    logging.CRITICAL: LogLevel.Fatal,
+    logging.ERROR: LogLevel.Error,
+    logging.WARNING: LogLevel.Warn,
+    logging.INFO: LogLevel.Info,
+    logging.DEBUG: LogLevel.Trace,
+}
+
+# Default Format supported by native loggers.
+CRT_LOG_FORMAT = '%(asctime)s [%(threadName)s] %(levelname)s %(name)s - %(message)s'
+
+
+class LogSubject(IntEnum):
+    """Log subject identifiers for CRT subsystems."""
+    # aws-c-common (package ID 0)
+    COMMON_GENERAL = 0x000
+    COMMON_TASK_SCHEDULER = 0x001
+
+    # aws-c-io (package ID 1)
+    IO_GENERAL = 0x400
+    IO_EVENT_LOOP = 0x401
+    IO_SOCKET = 0x402
+    IO_SOCKET_HANDLER = 0x403
+    IO_TLS = 0x404
+    IO_ALPN = 0x405
+    IO_DNS = 0x406
+    IO_PKI = 0x407
+    IO_CHANNEL = 0x408
+    IO_CHANNEL_BOOTSTRAP = 0x409
+    IO_FILE_UTILS = 0x40A
+    IO_SHARED_LIBRARY = 0x40B
+
+    # aws-c-http (package ID 2)
+    HTTP_GENERAL = 0x800
+    HTTP_CONNECTION = 0x801
+    HTTP_SERVER = 0x802
+    HTTP_STREAM = 0x803
+    HTTP_CONNECTION_MANAGER = 0x804
+    HTTP_WEBSOCKET = 0x805
+    HTTP_WEBSOCKET_SETUP = 0x806
+
+    # aws-c-event-stream (package ID 4)
+    EVENT_STREAM_GENERAL = 0x1000
+    EVENT_STREAM_CHANNEL_HANDLER = 0x1001
+    EVENT_STREAM_RPC_SERVER = 0x1002
+    EVENT_STREAM_RPC_CLIENT = 0x1003
+
+    # aws-c-mqtt (package ID 5)
+    MQTT_GENERAL = 0x1400
+    MQTT_CLIENT = 0x1401
+    MQTT_TOPIC_TREE = 0x1402
+
+    # aws-c-auth (package ID 6)
+    AUTH_GENERAL = 0x1800
+    AUTH_PROFILE = 0x1801
+    AUTH_CREDENTIALS_PROVIDER = 0x1802
+    AUTH_SIGNING = 0x1803
+
+    # aws-c-cal (package ID 7)
+    CAL_GENERAL = 0x1C00
+    CAL_ECC = 0x1C01
+    CAL_HASH = 0x1C02
+    CAL_HMAC = 0x1C03
+    CAL_DER = 0x1C04
+    CAL_LIBCRYPTO_RESOLVE = 0x1C05
+    CAL_RSA = 0x1C06
+    CAL_ED25519 = 0x1C07
+
+    # aws-c-s3 (package ID 14)
+    S3_GENERAL = 0x3800
+    S3_CLIENT = 0x3801
+
+    # aws-c-sdkutils (package ID 15)
+    SDKUTILS_GENERAL = 0x3C00
+    SDKUTILS_PROFILE = 0x3C01
+    SDKUTILS_ENDPOINTS_PARSING = 0x3C02
+    SDKUTILS_ENDPOINTS_RESOLVE = 0x3C03
+    SDKUTILS_ENDPOINTS_GENERAL = 0x3C04
+    SDKUTILS_PARTITIONS_PARSING = 0x3C05
+    SDKUTILS_ENDPOINTS_REGEX = 0x3C06
+
+
+_PACKAGE_ID_TO_MODULE = {
+    0: 'common',
+    1: 'io',
+    2: 'http',
+    4: 'event-stream',
+    5: 'mqtt',
+    6: 'auth',
+    7: 'cal',
+    14: 's3',
+    15: 'sdkutils',
+}
+
+
+def _python_logging_callback(crt_level, message, subject_id, subject_name, thread_name):
+    """Called from C for each CRT log message."""
+    module = _PACKAGE_ID_TO_MODULE.get(subject_id >> 10, 'unknown')
+    logger = logging.getLogger('awscrt.{}.{}'.format(module, subject_name))
+    py_level = _CRT_TO_PY_LEVEL.get(crt_level, logging.DEBUG)
+    # `There is the possibility that “dummy thread objects” are created. These are
+    # thread objects corresponding to “alien threads”, which are threads of control
+    # started outside the threading module, such as directly from C code.`
+    # https://docs.python.org/3/library/threading.html#thread-objects
+    #
+    # As per current conventions, dummy thread objects have thread names
+    # starting with "Dummy". Eg. Dummy-1, Dummy-8
+    if threading.current_thread().name.startswith("Dummy") and thread_name is not None:
+        threading.current_thread().name = thread_name
+    logger.log(py_level, '%s', message)
+
+
+def init_logging(level: int = logging.DEBUG):
+    """Initialize CRT logging, routing output through Python's logging module.
+
+    Log messages appear under the ``awscrt`` logger hierarchy, with each CRT
+    subsystem as a child logger (e.g. ``awscrt.io.event-loop``,
+    ``awscrt.s3.S3Client``).
+
+    This is mutually exclusive with :func:`~awscrt.io.init_logging` -- use one
+    or the other, not both. Can only be called once.
+
+    Example usage::
+
+        import logging
+        from awscrt.logging import init_logging
+
+        logging.basicConfig(level=logging.DEBUG)
+        init_logging(logging.DEBUG)
+
+    Args:
+        level (int): Python logging level (e.g. ``logging.DEBUG``,
+            ``logging.WARNING``).
+
+    Raises:
+        RuntimeError: If CRT logging has already been initialized.
+    """
+    root_logger = logging.getLogger('awscrt')
+
+    try:
+        crt_level = _PY_TO_CRT_LEVEL[level]
+    except KeyError:
+        raise ValueError(f"Invalid log level: {level}. Use logging.DEBUG, INFO, WARNING, ERROR, or CRITICAL")
+
+    if root_logger.level == logging.NOTSET:
+        root_logger.setLevel(_CRT_TO_PY_LEVEL.get(int(crt_level), logging.DEBUG))
+
+    _awscrt.init_python_logging(int(crt_level), _python_logging_callback)
+
+
+def set_log_level(level: int):
+    """Change the CRT log level. :func:`init_logging` must have been called first.
+
+    Set log level to logging.NOTSET to disable the logger. Cleaning up the logger
+    is dangerous in a multi-threaded environment.
+
+    Args:
+        level (int): Python logging level (e.g. ``logging.DEBUG``,
+            ``logging.WARNING``).
+    """
+    try:
+        crt_level = _PY_TO_CRT_LEVEL[level]
+    except KeyError:
+        raise ValueError(f"Invalid log level: {level}. Use logging.DEBUG, INFO, WARNING, ERROR, or CRITICAL")
+    _awscrt.set_log_level(int(crt_level))
+
+
+def log(level: int, subject: LogSubject, message: str):
+    """Log a message through the CRT's configured logger.
+
+    Args:
+        level (int): Python logging level (e.g. logging.DEBUG, logging.INFO).
+        subject (LogSubject): Log subject identifying the subsystem.
+        message (str): The message to log.
+    """
+    try:
+        crt_level = _PY_TO_CRT_LEVEL[level]
+    except KeyError:
+        raise ValueError(f"Invalid log level: {level}. Use logging.DEBUG, INFO, WARNING, ERROR, or CRITICAL")
+    _awscrt.logger_log(int(crt_level), int(subject), message)
diff --git a/source/io.h b/source/io.h
index fed29d856..818bcf62a 100644
--- a/source/io.h
+++ b/source/io.h
@@ -27,6 +27,7 @@ PyObject *aws_py_init_logging(PyObject *self, PyObject *args);
  * Change logging level of the global logger.
  */
 PyObject *aws_py_set_log_level(PyObject *self, PyObject *args);
+PyObject *aws_py_init_python_logging(PyObject *self, PyObject *args);
 
 /**
  * Returns True if ALPN is available, False if it is not.
diff --git a/source/module.c b/source/module.c
index 4eaa7b228..45f3322ea 100644
--- a/source/module.c
+++ b/source/module.c
@@ -31,10 +31,117 @@
 #include <aws/mqtt/mqtt.h>
 #include <aws/s3/s3.h>
 
+#include <aws/common/atomics.h>
+
 #include <memoryobject.h>
 
 static struct aws_logger s_logger;
-static bool s_logger_init = false;
+static struct aws_atomic_var s_logger_init;
+
+/* Python logging bridge: direct vtable override */
+struct s_py_logger_impl {
+    PyObject *callback;
+    struct aws_atomic_var level;
+};
+
+static int s_py_logger_log(
+    struct aws_logger *logger,
+    enum aws_log_level log_level,
+    aws_log_subject_t subject,
+    const char *format,
+    ...) {
+
+    struct s_py_logger_impl *impl = logger->p_impl;
+
+    va_list args;
+    va_start(args, format);
+    /* Calculate how much room we'll need to build the full log line.
+       You cannot consume a va_list twice, so we have to copy it. */
+    va_list tmp;
+    va_copy(tmp, args);
+    int len = vsnprintf(NULL, 0, format, tmp);
+    va_end(tmp);
+    if (len < 0) {
+        va_end(args);
+        return AWS_OP_ERR;
+    }
+
+    char *buf = aws_mem_acquire(logger->allocator, (size_t)len + 1);
+    vsnprintf(buf, (size_t)len + 1, format, args);
+    va_end(args);
+
+    const char *subject_name = aws_log_subject_name(subject);
+    if (!subject_name) {
+        subject_name = "";
+    }
+
+    struct aws_string *thread_name_str = NULL;
+    aws_thread_current_name(logger->allocator, &thread_name_str);
+
+    PyGILState_STATE state;
+    if (aws_py_gilstate_ensure(&state)) {
+        aws_string_destroy(thread_name_str);
+        aws_mem_release(logger->allocator, buf);
+        return AWS_OP_ERR;
+    }
+
+    PyObject *py_thread_name;
+
+    if (thread_name_str) {
+        py_thread_name = PyUnicode_FromAwsString(thread_name_str);
+    } else {
+        py_thread_name = Py_None;
+        Py_INCREF(py_thread_name);
+    }
+
+    aws_string_destroy(thread_name_str);
+
+    PyObject *result = PyObject_CallFunction(
+        impl->callback, "(is#isO)", (int)log_level, buf, (Py_ssize_t)len, (int)subject, subject_name, py_thread_name);
+    Py_XDECREF(result);
+    Py_DECREF(py_thread_name);
+    if (PyErr_Occurred()) {
+        PyErr_WriteUnraisable(PyErr_Occurred());
+        PyGILState_Release(state);
+        aws_mem_release(logger->allocator, buf);
+        return aws_py_raise_error();
+    }
+
+    PyGILState_Release(state);
+    aws_mem_release(logger->allocator, buf);
+    return AWS_OP_SUCCESS;
+}
+
+static enum aws_log_level s_py_logger_get_log_level(struct aws_logger *logger, aws_log_subject_t subject) {
+    (void)subject;
+    struct s_py_logger_impl *impl = logger->p_impl;
+    return (enum aws_log_level)aws_atomic_load_int(&impl->level);
+}
+
+static int s_py_logger_set_log_level(struct aws_logger *logger, enum aws_log_level level) {
+    struct s_py_logger_impl *impl = logger->p_impl;
+    aws_atomic_store_int(&impl->level, (size_t)level);
+    return AWS_OP_SUCCESS;
+}
+
+static void s_py_logger_clean_up(struct aws_logger *logger) {
+    struct s_py_logger_impl *impl = logger->p_impl;
+    if (impl) {
+        PyGILState_STATE state;
+        if (!aws_py_gilstate_ensure(&state)) {
+            Py_XDECREF(impl->callback);
+            PyGILState_Release(state);
+        }
+        aws_mem_release(logger->allocator, impl);
+    }
+}
+
+static struct aws_logger_vtable s_py_logger_vtable = {
+    .log = s_py_logger_log,
+    .get_log_level = s_py_logger_get_log_level,
+    .set_log_level = s_py_logger_set_log_level,
+    .clean_up = s_py_logger_clean_up,
+};
 
 PyObject *aws_py_init_logging(PyObject *self, PyObject *args) {
     (void)self;
@@ -70,18 +177,18 @@ PyObject *aws_py_init_logging(PyObject *self, PyObject *args) {
         log_options.filename = file_path;
     }
 
-    /* It is not safe to clean up a running logger */
-    if (s_logger_init) {
+    /* Atomically claim the init slot (0 → 1). Only one thread can win. */
+    size_t expected = 0;
+    if (!aws_atomic_compare_exchange_int(&s_logger_init, &expected, 1)) {
         aws_raise_error(AWS_ERROR_INVALID_STATE);
-        return PyErr_AwsLastError();
+        return PyErr_Format(PyExc_RuntimeError, "Invalid state: logger cannot be initialized multiple times.");
     }
 
-    if (aws_logger_init_standard(&s_logger, allocator, &log_options) == AWS_OP_SUCCESS) {
-        aws_logger_set(&s_logger);
-        s_logger_init = true;
-    } else {
+    if (aws_logger_init_standard(&s_logger, allocator, &log_options) != AWS_OP_SUCCESS) {
+        aws_atomic_store_int(&s_logger_init, 0); /* rollback on failure */
         return PyErr_AwsLastError();
     }
+    aws_logger_set(&s_logger);
 
     Py_RETURN_NONE;
 }
@@ -89,9 +196,9 @@ PyObject *aws_py_init_logging(PyObject *self, PyObject *args) {
 PyObject *aws_py_set_log_level(PyObject *self, PyObject *args) {
     (void)self;
 
-    if (!s_logger_init) {
+    if (!aws_atomic_load_int(&s_logger_init)) {
         aws_raise_error(AWS_ERROR_INVALID_STATE);
-        return PyErr_AwsLastError();
+        return PyErr_Format(PyExc_RuntimeError, "Invalid state: logger cannot be initialized multiple times.");
     }
 
     int log_level = 0;
@@ -107,6 +214,60 @@ PyObject *aws_py_set_log_level(PyObject *self, PyObject *args) {
     Py_RETURN_NONE;
 }
 
+PyObject *aws_py_init_python_logging(PyObject *self, PyObject *args) {
+    (void)self;
+
+    int log_level;
+    PyObject *py_callback;
+    if (!PyArg_ParseTuple(args, "iO", &log_level, &py_callback)) {
+        return NULL;
+    }
+
+    /* Atomically claim the init slot (0 → 1). Only one thread can win. */
+    size_t expected = 0;
+    if (!aws_atomic_compare_exchange_int(&s_logger_init, &expected, 1)) {
+        aws_raise_error(AWS_ERROR_INVALID_STATE);
+        return PyErr_AwsLastError();
+    }
+
+    if (!PyCallable_Check(py_callback)) {
+        aws_atomic_store_int(&s_logger_init, 0); /* rollback */
+        PyErr_SetString(PyExc_TypeError, "callback must be callable");
+        return NULL;
+    }
+
+    struct aws_allocator *allocator = aws_default_allocator();
+
+    struct s_py_logger_impl *impl = aws_mem_calloc(allocator, 1, sizeof(struct s_py_logger_impl));
+
+    Py_INCREF(py_callback);
+    impl->callback = py_callback;
+    aws_atomic_store_int(&impl->level, (size_t)log_level);
+
+    s_logger.vtable = &s_py_logger_vtable;
+    s_logger.allocator = allocator;
+    s_logger.p_impl = impl;
+
+    aws_logger_set(&s_logger);
+
+    Py_RETURN_NONE;
+}
+
+PyObject *aws_py_logger_log(PyObject *self, PyObject *args) {
+    (void)self;
+
+    int log_level;
+    int subject;
+    const char *message;
+    if (!PyArg_ParseTuple(args, "iis", &log_level, &subject, &message)) {
+        return NULL;
+    }
+
+    AWS_LOGF((enum aws_log_level)log_level, subject, "%s", message);
+
+    Py_RETURN_NONE;
+}
+
 struct aws_byte_cursor aws_byte_cursor_from_pyunicode(PyObject *str) {
     Py_ssize_t len;
     const char *ptr = PyUnicode_AsUTF8AndSize(str, &len);
@@ -789,6 +950,8 @@ static PyMethodDef s_module_methods[] = {
     AWS_PY_METHOD_DEF(tls_connection_options_set_server_name, METH_VARARGS),
     AWS_PY_METHOD_DEF(init_logging, METH_VARARGS),
     AWS_PY_METHOD_DEF(set_log_level, METH_VARARGS),
+    AWS_PY_METHOD_DEF(init_python_logging, METH_VARARGS),
+    AWS_PY_METHOD_DEF(logger_log, METH_VARARGS),
     AWS_PY_METHOD_DEF(input_stream_new, METH_VARARGS),
     AWS_PY_METHOD_DEF(pkcs11_lib_new, METH_VARARGS),
 
diff --git a/test/test_io.py b/test/test_io.py
index 6e414d4ba..20dd48178 100644
--- a/test/test_io.py
+++ b/test/test_io.py
@@ -2,8 +2,10 @@
 # SPDX-License-Identifier: Apache-2.0.
 
 from awscrt.io import *
+from awscrt.logging import init_logging, set_log_level, log, LogSubject
 from test import NativeResourceTest, TIMEOUT
 import io
+import logging
 import os
 import sys
 import unittest
@@ -248,5 +250,148 @@ def test_is_tls_cipher_supported(self):
         self.assertEqual(True, TlsCipherPref.DEFAULT.is_supported())
 
 
+class PythonLoggingTest(NativeResourceTest):
+    @classmethod
+    def setUpClass(cls):
+        init_logging(logging.DEBUG)
+
+    def test_log_level_and_message(self):
+        handler = logging.Handler()
+        handler.records = []
+        handler.emit = lambda record: handler.records.append(record)
+
+        logger = logging.getLogger('awscrt')
+        logger.setLevel(logging.DEBUG)
+        logger.addHandler(handler)
+
+        try:
+            log(logging.INFO, LogSubject.COMMON_GENERAL, "test message")
+
+            self.assertEqual(len(handler.records), 1)
+            self.assertEqual(handler.records[0].levelno, logging.INFO)
+            self.assertIn("test message", handler.records[0].getMessage())
+        finally:
+            logger.removeHandler(handler)
+
+    def test_log_subject(self):
+        handler = logging.Handler()
+        handler.records = []
+        handler.emit = lambda record: handler.records.append(record)
+
+        logger = logging.getLogger('awscrt')
+        logger.setLevel(logging.DEBUG)
+        logger.addHandler(handler)
+
+        try:
+            log(logging.DEBUG, LogSubject.IO_EVENT_LOOP, "event loop test")
+
+            self.assertEqual(len(handler.records), 1)
+            self.assertEqual(handler.records[0].name, "awscrt.io.event-loop")
+            self.assertIn("event loop test", handler.records[0].getMessage())
+        finally:
+            logger.removeHandler(handler)
+
+    def test_log_subject_hierarchy(self):
+        """Verify that log subjects from different CRT packages produce correct hierarchical logger names."""
+        cases = [
+            (LogSubject.COMMON_GENERAL, "awscrt.common."),
+            (LogSubject.IO_EVENT_LOOP, "awscrt.io."),
+            (LogSubject.HTTP_GENERAL, "awscrt.http."),
+            (LogSubject.EVENT_STREAM_GENERAL, "awscrt.event-stream."),
+            (LogSubject.MQTT_GENERAL, "awscrt.mqtt."),
+            (LogSubject.AUTH_GENERAL, "awscrt.auth."),
+            (LogSubject.CAL_GENERAL, "awscrt.cal."),
+            (LogSubject.S3_GENERAL, "awscrt.s3."),
+            (LogSubject.SDKUTILS_GENERAL, "awscrt.sdkutils."),
+        ]
+
+        handler = logging.Handler()
+        handler.records = []
+        handler.emit = lambda record: handler.records.append(record)
+
+        logger = logging.getLogger('awscrt')
+        logger.setLevel(logging.DEBUG)
+        logger.addHandler(handler)
+
+        try:
+            for subject, expected_prefix in cases:
+                handler.records.clear()
+                log(logging.DEBUG, subject, "test")
+                self.assertEqual(len(handler.records), 1)
+                self.assertTrue(
+                    handler.records[0].name.startswith(expected_prefix),
+                    f"Subject {subject.name}: expected logger starting with "
+                    f"'{expected_prefix}', got '{handler.records[0].name}'"
+                )
+        finally:
+            logger.removeHandler(handler)
+
+    def _emit_all_subjects_and_levels(self):
+        """Emit logs across all subjects at INFO and DEBUG levels."""
+        for subject in (LogSubject.IO_EVENT_LOOP, LogSubject.HTTP_GENERAL,
+                        LogSubject.EVENT_STREAM_GENERAL, LogSubject.MQTT_GENERAL,
+                        LogSubject.S3_GENERAL, LogSubject.AUTH_GENERAL,
+                        LogSubject.CAL_GENERAL, LogSubject.SDKUTILS_GENERAL,
+                        LogSubject.COMMON_GENERAL):
+            log(logging.INFO, subject, f"{subject.name} info")
+            log(logging.DEBUG, subject, f"{subject.name} debug")
+
+    def test_logging_filters(self):
+        """Verify level and subject filtering across different logger configurations."""
+        handler = logging.Handler()
+        handler.records = []
+        handler.emit = lambda record: handler.records.append(record)
+
+        root = logging.getLogger('awscrt')
+        root.addHandler(handler)
+
+        try:
+            # All at DEBUG: should get all 18 records (9 subjects x 2 levels)
+            root.setLevel(logging.DEBUG)
+            self._emit_all_subjects_and_levels()
+            self.assertEqual(len(handler.records), 18)
+            handler.records.clear()
+
+            # All at INFO: should get only 9 INFO records
+            root.setLevel(logging.INFO)
+            self._emit_all_subjects_and_levels()
+            self.assertEqual(len(handler.records), 9)
+            self.assertTrue(all(r.levelno == logging.INFO for r in handler.records))
+            handler.records.clear()
+
+            # All at WARNING: nothing should pass
+            root.setLevel(logging.WARNING)
+            self._emit_all_subjects_and_levels()
+            self.assertEqual(len(handler.records), 0)
+
+            # Enable only HTTP at DEBUG: should get 2 HTTP records
+            logging.getLogger('awscrt.http').setLevel(logging.DEBUG)
+            self._emit_all_subjects_and_levels()
+            self.assertEqual(len(handler.records), 2)
+            self.assertTrue(all(r.name.startswith("awscrt.http.") for r in handler.records))
+            handler.records.clear()
+
+            # Also enable S3 at INFO: should get 2 HTTP + 1 S3 INFO
+            logging.getLogger('awscrt.s3').setLevel(logging.INFO)
+            self._emit_all_subjects_and_levels()
+            http_records = [r for r in handler.records if r.name.startswith("awscrt.http.")]
+            s3_records = [r for r in handler.records if r.name.startswith("awscrt.s3.")]
+            self.assertEqual(len(http_records), 2)
+            self.assertEqual(len(s3_records), 1)
+            self.assertEqual(s3_records[0].levelno, logging.INFO)
+            self.assertEqual(len(handler.records), 3)
+            handler.records.clear()
+
+        finally:
+            root.setLevel(logging.DEBUG)
+            logging.getLogger('awscrt.http').setLevel(logging.NOTSET)
+            logging.getLogger('awscrt.s3').setLevel(logging.NOTSET)
+            root.removeHandler(handler)
+
+    @classmethod
+    def tearDownClass(cls):
+        set_log_level(logging.NOTSET)
+
+
 if __name__ == '__main__':
     unittest.main()