more

src/prefect/locking/filesystem.py

@@ -4,12 +4,12 @@
 from typing import Optional
 
 import anyio
-import pendulum
 import pydantic_core
 from typing_extensions import TypedDict
 
 from prefect.logging.loggers import get_logger
 from prefect.types import DateTime
+from prefect.types._datetime import Duration, parse_datetime
 
 from .protocol import LockManager
 
@@ -64,7 +64,7 @@ def _get_lock_info(self, key: str, use_cache: bool = True) -> Optional[_LockInfo
                 lock_info["path"] = lock_path
                 expiration = lock_info.get("expiration")
                 lock_info["expiration"] = (
-                    pendulum.parse(expiration) if expiration is not None else None
+                    parse_datetime(expiration) if expiration is not None else None
                 )
             self._locks[key] = lock_info
             return lock_info
@@ -86,7 +86,7 @@ async def _aget_lock_info(
             lock_info["path"] = lock_path
             expiration = lock_info.get("expiration")
             lock_info["expiration"] = (
-                pendulum.parse(expiration) if expiration is not None else None
+                parse_datetime(expiration) if expiration is not None else None
             )
             self._locks[key] = lock_info
             return lock_info
@@ -117,7 +117,7 @@ def acquire_lock(
                 )
                 return self.acquire_lock(key, holder, acquire_timeout, hold_timeout)
         expiration = (
-            DateTime.now("utc") + pendulum.duration(seconds=hold_timeout)
+            DateTime.now("utc") + Duration(seconds=hold_timeout)
             if hold_timeout is not None
             else None
         )
@@ -166,7 +166,7 @@ async def aacquire_lock(
                 )
                 return self.acquire_lock(key, holder, acquire_timeout, hold_timeout)
         expiration = (
-            DateTime.now("utc") + pendulum.duration(seconds=hold_timeout)
+            DateTime.now("utc") + Duration(seconds=hold_timeout)
             if hold_timeout is not None
             else None
         )

src/prefect/logging/handlers.py

@@ -33,6 +33,7 @@
     PREFECT_LOGGING_TO_API_MAX_LOG_SIZE,
     PREFECT_LOGGING_TO_API_WHEN_MISSING_FLOW,
 )
+from prefect.types._datetime import from_timestamp
 
 
 class APILogWorker(BatchedQueueService[Dict[str, Any]]):
@@ -219,9 +220,7 @@ def prepare(self, record: logging.LogRecord) -> Dict[str, Any]:
             worker_id=worker_id,
             name=record.name,
             level=record.levelno,
-            timestamp=pendulum.from_timestamp(
-                getattr(record, "created", None) or time.time()
-            ),
+            timestamp=from_timestamp(getattr(record, "created", None) or time.time()),
             message=self.format(record),
         ).model_dump(mode="json")
 

src/prefect/types/_datetime.py

@@ -2,7 +2,21 @@
 
 from typing import Annotated, Any
 
-from pendulum import FixedTimezone, format_diff, from_format, instance, parse, tz
+from pendulum import (
+    FixedTimezone,
+    format_diff,
+    parse,
+    tz,
+)
+from pendulum import (
+    from_format as from_format_pendulum,
+)
+from pendulum import (
+    from_timestamp as from_timestamp_pendulum,
+)
+from pendulum import (
+    instance as instance_pendulum,
+)
 from pendulum.tz import UTC, Timezone
 from pydantic import Field
 from pydantic_extra_types.pendulum_dt import Date as PydanticDate
@@ -32,21 +46,30 @@ def datetime_from_format(
     tz: str | Timezone = UTC,
     locale: str | None = None,
 ) -> DateTime:
-    return PydanticDateTime(from_format(string, fmt, tz, locale))  # type: ignore
+    return from_format_pendulum(string, fmt, tz, locale)  # type: ignore
 
 
 def parse_datetime(string: str) -> DateTime:
     return parse(string)  # type: ignore
 
 
 def datetime_instance(value: Any) -> DateTime:
-    return instance(value)  # type: ignore
+    """
+    Here to fulfull the needs that pendulum.instance meets, but it
+    is often used ambiguously and we should phase this out over time
+    in favor of more explicit datetime utilities.
+    """
+    return instance_pendulum(value)  # type: ignore
 
 
 def local_timezone() -> Timezone | FixedTimezone:
     return tz.local_timezone()
 
 
+def from_timestamp(timestamp: float) -> DateTime:
+    return from_timestamp_pendulum(timestamp)  # type: ignore
+
+
 __all__ = [
     "DateTime",
     "Date",

tests/docker/test_registry_pushes.py

@@ -31,7 +31,7 @@ def contexts() -> Path:
 @pytest.fixture(scope="module")
 def frozen_now():
     now = DateTime.now("UTC")
-    with mock.patch("DateTime.now", return_value=now):
+    with mock.patch("prefect.types.DateTime.now", return_value=now):
         yield now
 
 
@@ -45,8 +45,7 @@ def howdy(docker: DockerClient, worker_id: str, frozen_now: DateTime) -> str:
         image.add_line(f'ENTRYPOINT [ "echo", "{message}" ]')
         image_id = image.build()
 
-    greeting = docker.containers.run(image_id, remove=True).decode().strip()
-    assert greeting == message
+    assert docker.containers.run(image_id, remove=True).decode().strip() == message  # type: ignore
 
     # Give the image a unit tag for this run we we can confirm it is only untagged but
     # not removed by the process of pushing it to the registry