Parquet metadata persistence of DataFrame.attrs

pandas/io/parquet.py

@@ -1,7 +1,9 @@
 """ parquet compat """
 from __future__ import annotations
 
+import ast
 import io
+import json
 import os
 from typing import (
     TYPE_CHECKING,
@@ -184,6 +186,11 @@ def write(
 
         table = self.api.Table.from_pandas(df, **from_pandas_kwargs)
 
+        df_metadata = {"df.attrs": json.dumps(df.attrs)}
+        existing_metadata = table.schema.metadata
+        merged_metadata = {**existing_metadata, **df_metadata}
+        table = table.replace_schema_metadata(merged_metadata)
+
         path_or_handle, handles, filesystem = _get_path_or_handle(
             path,
             filesystem,
@@ -263,6 +270,11 @@ def read(
 
             if manager == "array":
                 result = result._as_manager("array", copy=False)
+
+            if pa_table.schema.metadata:
+                if b"df.attrs" in pa_table.schema.metadata:
+                    df_metadata = pa_table.schema.metadata[b"df.attrs"]
+                    result.attrs = ast.literal_eval(df_metadata.decode("utf-8"))
             return result
         finally:
             if handles is not None:

pandas/tests/io/test_parquet.py

@@ -1065,6 +1065,14 @@ def test_empty_columns(self, pa):
         df = pd.DataFrame(index=pd.Index(["a", "b", "c"], name="custom name"))
         check_round_trip(df, pa)
 
+    def test_df_attrs_persistence(self, tmp_path, pa):
+        path = tmp_path / "test_df_metadata.p"
+        df = pd.DataFrame(data={1: [1]})
+        df.attrs = {"test_attribute": 1}
+        df.to_parquet(path, engine=pa)
+        new_df = read_parquet(path, engine=pa)
+        assert new_df.attrs == df.attrs
+
 
 class TestParquetFastParquet(Base):
     def test_basic(self, fp, df_full):