fix_email_error_add_abfs_features

deanm0000 · deanm0000 · commit f13a0042824b · 2025-07-01T10:39:22.000-04:00
diff --git a/pyproject.toml b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dean_utils"
-version="0.0.51"
+version="0.0.52"
 authors=[
     { name="Dean MacGregor", email="powertrading121@gmail.com"}
 ]
@@ -80,7 +80,7 @@ ignore = [
   # Lints below are turned off because of conflicts with the ruff formatter
   "D206",
   "W191",
-  "D100", "D101", "D102", "D103"
+  "D100", "D101", "D102", "D103", "SIM112"
 ]
 
 [tool.ruff.lint.pycodestyle]
diff --git a/requirements.txt b/requirements.txt
@@ -6,4 +6,5 @@ fsspec
 azure-storage-queue
 python-multipart
 azure-communication-email
-httpx
+httpx
+aiopath
diff --git a/src/dean_utils/__init__.py b/src/dean_utils/__init__.py
@@ -1,22 +1,19 @@
 __all__ = [
     "async_abfs",
+    "az_send",
     "clear_messages",
-    "peek_messages",
-    "get_queue_properties",
-    "send_message",
-    "update_queue",
     "delete_message",
-    "send_email",
-    "az_send",
+    "get_queue_properties",
+    "global_async_client",
+    "peek_messages",
     "pl_scan_hive",
     "pl_scan_pq",
-    "pl_write_pq",
     "pl_write_delta_append",
-    "global_async_client",
+    "pl_write_pq",
+    "send_message",
+    "update_queue",
 ]
 import contextlib
-from collections.abc import Iterable
-from typing import cast
 
 from dean_utils.polars_extras import (
     pl_scan_hive,
@@ -27,6 +24,8 @@
 with contextlib.suppress(ImportError):
     from dean_utils.polars_extras import pl_write_delta_append
 
+from pathlib import Path
+
 from dean_utils.utils.az_utils import (
     async_abfs,
     clear_messages,
@@ -36,33 +35,33 @@
     send_message,
     update_queue,
 )
-from dean_utils.utils.email_utility import az_send, send_email
+from dean_utils.utils.email_utility import az_send
 from dean_utils.utils.httpx import global_async_client
 
 
 def error_email(func, attempts=1):
     def wrapper(*args, **kwargs):
-        subject = None
+        subject = Path.cwd()
         errors = []
         for _ in range(attempts):
             try:
                 return func(*args, **kwargs)
             except Exception as err:
                 import inspect
-                from pathlib import Path
                 from traceback import format_exception
 
-                if subject is None:
-                    subject = Path.cwd()
-                errors.append(
-                    "\n".join(cast(Iterable[str], inspect.stack()))
-                    + "\n\n"
-                    + "\n".join(format_exception(err))
+                filt_stack = "\n".join(
+                    [
+                        str(x)
+                        for x in inspect.stack()[1:]
+                        if "site-packages" not in x.filename
+                    ]
                 )
-        if subject is not None:
-            az_send(
-                str(subject),
-                "\n".join(errors),
-            )
+                errors.append("\n".join(["\n".join(format_exception(err)), filt_stack]))
+
+        az_send(
+            str(subject),
+            "\n".join(errors),
+        )
 
     return wrapper
diff --git a/src/dean_utils/utils/az_utils.py b/src/dean_utils/utils/az_utils.py
@@ -10,6 +10,7 @@
 
 import azure.storage.blob as asb
 import fsspec
+from aiopath import AsyncPath
 from azure.core.exceptions import HttpResponseError
 from azure.storage.blob import BlobBlock
 from azure.storage.blob.aio import BlobClient
@@ -18,10 +19,11 @@
 
 if TYPE_CHECKING:
     import httpx
+    from azure.storage.blob.aio import StorageStreamDownloader
     from azure.storage.queue import QueueMessage
 
 HTTPX_METHODS: TypeAlias = Literal["GET", "POST"]
-AIO_SERVE = QSC.from_connection_string(conn_str=os.environ["AzureWebJobsStorage"])  # noqa: SIM112
+AIO_SERVE = QSC.from_connection_string(conn_str=os.environ["AzureWebJobsStorage"])
 
 
 async def peek_messages(queue: str, max_messages: int | None = None, **kwargs):
@@ -149,7 +151,7 @@ async def clear_messages(
 
 
 class async_abfs:
-    def __init__(self, connection_string=os.environ["Synblob"]):  # noqa: SIM112
+    def __init__(self, connection_string=os.environ["Synblob"]):
         self.connection_string = connection_string
         self.sync = fsspec.filesystem("abfss", connection_string=self.connection_string)
         key_conv = {"AccountName": "account_name", "AccountKey": "account_key"}
@@ -189,13 +191,12 @@ async def stream_dl(
         **httpx_extras,
     ) -> None:
         """
-        Help on method stream_dl.
+        stream_dl will stream the contents of a url to a path in the cloud given an httpx Client.
 
-        async stream_dl(client, method, url, **httpx_extras)
+        async stream_dl(client, method, url, path, recurs, **httpx_extras)
             Download file streaming in chunks in async as downloader and to a Blob
 
-        Parameters
-        ----------
+        Args:
             client: httpx.AsyncClient
                 The httpx Async Client object to use
             method:
@@ -204,7 +205,9 @@ async def stream_dl(
                 The URL to download
             path:
                 The full path to Azure file being saved
-            **httpx_extras
+            recurs:
+                To try again recursively
+            httpx_extras
                 Any extra arguments to be sent to client.stream
         """
         async with (
@@ -216,7 +219,7 @@ async def stream_dl(
             resp.raise_for_status()
             block_list = []
             async for chunk in resp.aiter_bytes():
-                chunk = cast(IO, chunk)
+                chunk = cast("IO", chunk)
                 block_id = uuid4().hex
                 try:
                     await target.stage_block(block_id=block_id, data=chunk)
@@ -244,60 +247,63 @@ async def stream_dl(
 
     async def stream_up(
         self,
-        local_path: str | Path,
+        local_path: str | Path | AsyncPath,
         remote_path: str,
         size: int = 16384,
         /,
         recurs=False,
     ) -> None:
         """
-        Help on method stream_dl.
+        Help on method stream_up.
 
-        async stream_dl(client, method, url, **httpx_extras)
+        async stream_up(local_path, remote_path, size, recurs)
             Download file streaming in chunks in async as downloader and to a Blob
 
-        Parameters
-        ----------
+        Args:
             local_path:
                 The full path to local path as str or Path
             remote_path:
                 The full path to remote path as str
             size:
                 The number of bytes read per iteration in read
+            recurs:
+                To try again recursively
         """
-        if isinstance(local_path, str):
-            local_path = Path(local_path)
-        async with BlobClient.from_connection_string(
-            self.connection_string, *(remote_path.split("/", maxsplit=1))
-        ) as target:
-            with local_path.open("rb") as src:
-                block_list = []
-                while True:
-                    chunk = src.read(size)
-                    chunk = cast(IO, chunk)
-                    if not chunk:
-                        break
-                    block_id = uuid4().hex
-                    try:
-                        await target.stage_block(block_id=block_id, data=chunk)
-                    except HttpResponseError as err:
-                        if "The specified blob or block content is invalid." not in str(
-                            err
-                        ):
-                            raise
-                        await asyncio.sleep(1)
-                        await target.commit_block_list([])
-                        await target.delete_blob()
-                        if recurs is False:
-                            await self.stream_up(
-                                local_path,
-                                remote_path,
-                                recurs=True,
-                            )
-                        else:
-                            raise
-                    block_list.append(BlobBlock(block_id=block_id))
-                await target.commit_block_list(block_list)
+        if isinstance(local_path, (str, Path)):
+            local_path = AsyncPath(local_path)
+        async with (
+            BlobClient.from_connection_string(
+                self.connection_string, *(remote_path.split("/", maxsplit=1))
+            ) as target,
+            local_path.open("rb") as src,
+        ):
+            block_list = []
+            while True:
+                chunk = await src.read(size)
+                chunk = cast("IO", chunk)
+                if not chunk:
+                    break
+                block_id = uuid4().hex
+                try:
+                    await target.stage_block(block_id=block_id, data=chunk)
+                except HttpResponseError as err:
+                    if "The specified blob or block content is invalid." not in str(
+                        err
+                    ):
+                        raise
+                    await asyncio.sleep(1)
+                    await target.commit_block_list([])
+                    await target.delete_blob()
+                    if recurs is False:
+                        await self.stream_up(
+                            local_path,
+                            remote_path,
+                            recurs=True,
+                        )
+                    else:
+                        raise
+                block_list.append(BlobBlock(block_id=block_id))
+            await target.commit_block_list(block_list)
 
     async def walk(self, path: str, maxdepth=None, **kwargs):
         """
@@ -312,16 +318,16 @@ async def walk(self, path: str, maxdepth=None, **kwargs):
             Note that the "files" outputted will include anything that is not
             a directory, such as links.
 
-        Parameters
-        ----------
+        Args:
             path: str
                 Root to recurse into
 
             maxdepth: int
                 Maximum recursion depth. None means limitless, but not recommended
                 on link-based file-systems.
 
-            **kwargs are passed to ``ls``
+            kwargs:
+                dict of args passed to ``ls``
         """
         this_fs = fsspec.filesystem(
             "abfss", connection_string=self.connection_string, asyncronous=True
@@ -359,8 +365,7 @@ async def details(
             AzureBlobFileSystem instance
             Return a list of dictionaries of specifying details about the contents
 
-        Parameters
-        ----------
+        Args:
             contents
 
             delimiter: str
@@ -442,8 +447,7 @@ async def ls(
         versions: bool = False, **kwargs) method of adlfs.spec.AzureBlobFileSystem instance
             Create a list of blob names from a blob container
 
-        Parameters
-        ----------
+        Args:
             path: str
                 Path to an Azure Blob with its container name
 
@@ -486,8 +490,7 @@ async def rm(
         """
         Delete files.
 
-        Parameters
-        ----------
+        Args:
         path: str or list of str
             File(s) to delete.
         recursive: bool
@@ -538,3 +541,10 @@ def make_sas_link(self, filepath, expiry=None, write=False):
             expiry=expiry,
         )
         return f"https://{account_dict['AccountName']}.blob.core.windows.net/{filepath}?{sas}"
+
+    async def stream(self, path: str) -> StorageStreamDownloader[bytes]:
+        blob = BlobClient.from_connection_string(
+            self.connection_string, *(path.split("/", maxsplit=1))
+        )
+        stream = await blob.download_blob()
+        return stream
diff --git a/src/dean_utils/utils/email_utility.py b/src/dean_utils/utils/email_utility.py
