diff --git a/lang/py/avro/io.py b/lang/py/avro/io.py
index f5063a6686f..16f79c5e4de 100644
--- a/lang/py/avro/io.py
+++ b/lang/py/avro/io.py
@@ -558,6 +558,7 @@ def write_decimal_bytes(self, datum: decimal.Decimal, scale: int) -> None:
         for digit in digits:
             unscaled_datum = (unscaled_datum * 10) + digit
 
+        unscaled_datum *= 10 ** (scale + int(exp))
         bits_req = unscaled_datum.bit_length() + 1
         if sign:
             unscaled_datum = (1 << bits_req) - unscaled_datum
@@ -584,6 +585,7 @@ def write_decimal_fixed(self, datum: decimal.Decimal, scale: int, size: int) ->
         for digit in digits:
             unscaled_datum = (unscaled_datum * 10) + digit
 
+        unscaled_datum *= 10 ** (scale + int(exp))
         bits_req = unscaled_datum.bit_length() + 1
         size_in_bits = size * 8
         offset_bits = size_in_bits - bits_req
diff --git a/lang/py/avro/test/test_io.py b/lang/py/avro/test/test_io.py
index 1a71e0598e5..4acdcab9783 100644
--- a/lang/py/avro/test/test_io.py
+++ b/lang/py/avro/test/test_io.py
@@ -756,6 +756,37 @@ def test_recursive_record_min_bytes_is_non_zero(self) -> None:
 
 
 class TestMisc(unittest.TestCase):
+    def test_decimal_encoding_preserves_value_at_schema_scale(self) -> None:
+        cases = (
+            ("3.1", 2, 310),
+            ("-3.1", 2, -310),
+            ("3", 2, 300),
+            ("-3", 2, -300),
+            ("3E+1", 2, 3000),
+            ("-3E+1", 2, -3000),
+            ("3E+1", 0, 30),
+            ("3.1", 4, 31000),
+            ("3.14", 2, 314),
+            ("0", 2, 0),
+            ("12345.6", 2, 1234560),
+        )
+        for storage_type, (value, scale, expected_unscaled) in itertools.product(("bytes", "fixed"), cases):
+            with self.subTest(storage_type=storage_type, value=value, scale=scale):
+                schema = avro.schema.parse(
+                    json.dumps({"type": storage_type, "logicalType": "decimal", "precision": 8, "scale": scale, "name": "Test", "size": 4})
+                )
+                datum = decimal.Decimal(value)
+                # Scaling must not round according to the caller's decimal context.
+                with decimal.localcontext() as context:
+                    context.prec = 2
+                    writer, _, _ = write_datum(datum, schema)
+                encoded = writer.getvalue()
+                decoder = avro.io.BinaryDecoder(io.BytesIO(encoded))
+                payload = decoder.read_bytes() if storage_type == "bytes" else decoder.read(4)
+                self.assertEqual(int.from_bytes(payload, byteorder="big", signed=True), expected_unscaled)
+                decoded = avro.io.DatumReader(schema).read(avro.io.BinaryDecoder(io.BytesIO(encoded)))
+                self.assertEqual(decoded, datum)
+
     def test_decimal_bytes_small_scale(self) -> None:
         """Avro should raise an AvroTypeException when attempting to write a decimal with a larger exponent than the schema's scale."""
         datum = decimal.Decimal("3.1415")