Replace deprecated datetime.utcnow() with datetime.now(timezone.utc)

docs/book/how-to/pipeline-development/develop-locally/keep-your-dashboard-server-clean.md

@@ -59,7 +59,7 @@ def delete_recent_pipeline_runs():
     zc = Client()
 
     # Calculate the timestamp for 24 hours ago
-    twenty_four_hours_ago = datetime.datetime.utcnow() - datetime.timedelta(hours=24)
+    twenty_four_hours_ago = datetime.datetime.now(timezone.utc) - datetime.timedelta(hours=24)
 
     # Format the timestamp as required by ZenML
     time_filter = twenty_four_hours_ago.strftime("%Y-%m-%d %H:%M:%S")

src/zenml/cli/service_connectors.py

@@ -13,7 +13,7 @@
 #  permissions and limitations under the License.
 """Service connector CLI commands."""
 
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional, Union, cast
 from uuid import UUID
 
@@ -291,7 +291,9 @@ def prompt_expires_at(
     while True:
         default_str = ""
         if default is not None:
-            seconds = int((default - datetime.utcnow()).total_seconds())
+            seconds = int(
+                (default - datetime.now(timezone.utc)).total_seconds()
+            )
             default_str = (
                 f" [{str(default)} i.e. in "
                 f"{seconds_to_human_readable(seconds)}]"
@@ -307,14 +309,16 @@ def prompt_expires_at(
 
         assert expires_at is not None
         assert isinstance(expires_at, datetime)
-        if expires_at < datetime.utcnow():
+        if expires_at < datetime.now(timezone.utc):
             cli_utils.warning(
                 "The expiration time must be in the future. Please enter a "
                 "later date and time."
             )
             continue
 
-        seconds = int((expires_at - datetime.utcnow()).total_seconds())
+        seconds = int(
+            (expires_at - datetime.now(timezone.utc)).total_seconds()
+        )
 
         confirm = click.confirm(
             f"Credentials will be valid until {str(expires_at)} UTC (i.e. "

src/zenml/cli/stack.py

@@ -17,7 +17,7 @@
 import re
 import time
 import webbrowser
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -1576,7 +1576,7 @@ def deploy(
         ):
             raise click.Abort()
 
-        date_start = datetime.utcnow()
+        date_start = datetime.now(timezone.utc)
 
         webbrowser.open(deployment_config.deployment_url)
         console.print(

src/zenml/config/pipeline_configurations.py

@@ -13,7 +13,7 @@
 #  permissions and limitations under the License.
 """Pipeline configuration classes."""
 
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 from pydantic import SerializeAsAny, field_validator
@@ -61,7 +61,7 @@ def _get_full_substitutions(
             The full substitutions dict including date and time.
         """
         if start_time is None:
-            start_time = datetime.utcnow()
+            start_time = datetime.now(timezone.utc)
         ret = self.substitutions.copy()
         ret.setdefault("date", start_time.strftime("%Y_%m_%d"))
         ret.setdefault("time", start_time.strftime("%H_%M_%S_%f"))

src/zenml/event_hub/base_event_hub.py

@@ -14,7 +14,7 @@
 """Base class for event hub implementations."""
 
 from abc import ABC, abstractmethod
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple
 
 from zenml import EventSourceResponse
@@ -134,7 +134,7 @@ def trigger_action(
         )
         expires: Optional[datetime] = None
         if trigger.action.auth_window:
-            expires = datetime.utcnow() + timedelta(
+            expires = datetime.now(timezone.utc) + timedelta(
                 minutes=trigger.action.auth_window
             )
         encoded_token = token.encode(expires=expires)

src/zenml/integrations/airflow/orchestrators/airflow_orchestrator.py

@@ -408,7 +408,8 @@ def _translate_schedule(
         if schedule:
             if schedule.cron_expression:
                 start_time = schedule.start_time or (
-                    datetime.datetime.utcnow() - datetime.timedelta(7)
+                    datetime.datetime.now(datetime.timezone.utc)
+                    - datetime.timedelta(7)
                 )
                 return {
                     "schedule": schedule.cron_expression,
@@ -428,6 +429,7 @@ def _translate_schedule(
             "schedule": "@once",
             # set a start time in the past and disable catchup so airflow
             # runs the dag immediately
-            "start_date": datetime.datetime.utcnow() - datetime.timedelta(7),
+            "start_date": datetime.datetime.now(datetime.timezone.utc)
+            - datetime.timedelta(7),
             "catchup": False,
         }

src/zenml/integrations/kubernetes/orchestrators/kube_utils.py

@@ -248,7 +248,7 @@ def wait_pod(
     Returns:
         The pod object which meets the exit condition.
     """
-    start_time = datetime.datetime.utcnow()
+    start_time = datetime.datetime.now(datetime.timezone.utc)
 
     # Link to exponential back-off algorithm used here:
     # https://cloud.google.com/storage/docs/exponential-backoff
@@ -288,7 +288,7 @@ def wait_pod(
             return resp
 
         # Check if wait timed out.
-        elapse_time = datetime.datetime.utcnow() - start_time
+        elapse_time = datetime.datetime.now(datetime.timezone.utc) - start_time
         if elapse_time.seconds >= timeout_sec and timeout_sec != 0:
             raise RuntimeError(
                 f"Waiting for pod `{namespace}:{pod_name}` timed out after "

src/zenml/integrations/whylogs/data_validators/whylogs_data_validator.py

@@ -97,7 +97,9 @@ def data_profiling(
         """
         results = why.log(pandas=dataset)
         profile = results.profile()
-        dataset_timestamp = dataset_timestamp or datetime.datetime.utcnow()
+        dataset_timestamp = dataset_timestamp or datetime.datetime.now(
+            datetime.timezone.utc
+        )
         profile.set_dataset_timestamp(dataset_timestamp=dataset_timestamp)
         return profile.view()
 

src/zenml/models/v2/core/api_key.py

@@ -13,7 +13,7 @@
 #  permissions and limitations under the License.
 """Models representing API keys."""
 
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, ClassVar, List, Optional, Type, Union
 from uuid import UUID
 
@@ -319,7 +319,7 @@ def verify_key(
             and self.retain_period_minutes > 0
         ):
             # check if the previous key is still valid
-            if datetime.utcnow() - self.last_rotated < timedelta(
+            if datetime.now(timezone.utc) - self.last_rotated < timedelta(
                 minutes=self.retain_period_minutes
             ):
                 key_hash = self.previous_key

src/zenml/orchestrators/publish_utils.py

@@ -13,7 +13,7 @@
 #  permissions and limitations under the License.
 """Utilities to publish pipeline and step runs."""
 
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Dict, List
 
 from zenml.client import Client
@@ -48,7 +48,7 @@ def publish_successful_step_run(
         step_run_id=step_run_id,
         step_run_update=StepRunUpdate(
             status=ExecutionStatus.COMPLETED,
-            end_time=datetime.utcnow(),
+            end_time=datetime.now(timezone.utc),
             outputs=output_artifact_ids,
         ),
     )
@@ -67,7 +67,7 @@ def publish_failed_step_run(step_run_id: "UUID") -> "StepRunResponse":
         step_run_id=step_run_id,
         step_run_update=StepRunUpdate(
             status=ExecutionStatus.FAILED,
-            end_time=datetime.utcnow(),
+            end_time=datetime.now(timezone.utc),
         ),
     )
 
@@ -87,7 +87,7 @@ def publish_failed_pipeline_run(
         run_id=pipeline_run_id,
         run_update=PipelineRunUpdate(
             status=ExecutionStatus.FAILED,
-            end_time=datetime.utcnow(),
+            end_time=datetime.now(timezone.utc),
         ),
     )
 

src/zenml/orchestrators/step_launcher.py

@@ -16,7 +16,7 @@
 import os
 import time
 from contextlib import nullcontext
-from datetime import datetime
+from datetime import datetime, timezone
 from functools import partial
 from typing import TYPE_CHECKING, Any, Callable, Dict, Tuple
 
@@ -201,7 +201,7 @@ def launch(self) -> None:
                         f"Failed preparing step `{self._step_name}`."
                     )
                     step_run_request.status = ExecutionStatus.FAILED
-                    step_run_request.end_time = datetime.utcnow()
+                    step_run_request.end_time = datetime.now(timezone.utc)
                     raise
                 finally:
                     step_run = Client().zen_store.create_run_step(
@@ -305,7 +305,7 @@ def _create_or_reuse_run(self) -> Tuple[PipelineRunResponse, bool]:
             The created or existing pipeline run,
             and a boolean indicating whether the run was created or reused.
         """
-        start_time = datetime.utcnow()
+        start_time = datetime.now(timezone.utc)
         run_name = string_utils.format_name_template(
             name_template=self._deployment.run_name_template,
             substitutions=self._deployment.pipeline_configuration._get_full_substitutions(

src/zenml/orchestrators/step_run_utils.py

@@ -13,7 +13,7 @@
 #  permissions and limitations under the License.
 """Utilities for creating step runs."""
 
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Dict, List, Optional, Set, Tuple
 
 from zenml.client import Client
@@ -75,7 +75,7 @@ def create_request(self, invocation_id: str) -> StepRunRequest:
             pipeline_run_id=self.pipeline_run.id,
             deployment=self.deployment.id,
             status=ExecutionStatus.RUNNING,
-            start_time=datetime.utcnow(),
+            start_time=datetime.now(timezone.utc),
             user=Client().active_user.id,
             workspace=Client().active_workspace.id,
         )

src/zenml/pipelines/run_utils.py

@@ -1,7 +1,7 @@
 """Utility functions for running pipelines."""
 
 import time
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union
 from uuid import UUID
 
@@ -65,7 +65,7 @@ def create_placeholder_run(
 
     if deployment.schedule:
         return None
-    start_time = datetime.utcnow()
+    start_time = datetime.now(timezone.utc)
     run_request = PipelineRunRequest(
         name=string_utils.format_name_template(
             name_template=deployment.run_name_template,

src/zenml/service_connectors/service_connector.py

@@ -799,8 +799,8 @@ def to_response_model(
             name=name,
             body=ServiceConnectorResponseBody(
                 user=user,
-                created=datetime.utcnow(),
-                updated=datetime.utcnow(),
+                created=datetime.now(timezone.utc),
+                updated=datetime.now(timezone.utc),
                 description=description,
                 connector_type=self.get_type(),
                 auth_method=self.auth_method,

src/zenml/stack/stack.py

@@ -16,7 +16,7 @@
 import itertools
 import json
 import os
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import (
     TYPE_CHECKING,
     AbstractSet,
@@ -751,8 +751,8 @@ def validate_image_builder(self) -> None:
                 config=LocalImageBuilderConfig(),
                 user=Client().active_user.id,
                 workspace=Client().active_workspace.id,
-                created=datetime.utcnow(),
-                updated=datetime.utcnow(),
+                created=datetime.now(timezone.utc),
+                updated=datetime.now(timezone.utc),
             )
 
             self._image_builder = image_builder

src/zenml/utils/git_utils.py

@@ -16,7 +16,7 @@
 import os
 from typing import Optional
 
-from git import GitCommandError
+from git.exc import GitCommandError
 from git.repo import Repo
 
 

src/zenml/utils/string_utils.py

@@ -17,7 +17,7 @@
 import functools
 import random
 import string
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Any, Callable, Dict, Optional, TypeVar, cast
 
 from pydantic import BaseModel
@@ -180,7 +180,7 @@ def format_name_template(
             start_time = None
 
         if start_time is None:
-            start_time = datetime.utcnow()
+            start_time = datetime.now(timezone.utc)
         substitutions.setdefault("date", start_time.strftime("%Y_%m_%d"))
         substitutions.setdefault("time", start_time.strftime("%H_%M_%S_%f"))
 

src/zenml/zen_server/auth.py

@@ -14,7 +14,7 @@
 """Authentication module for ZenML server."""
 
 from contextvars import ContextVar
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Callable, Optional, Union
 from urllib.parse import urlencode
 from uuid import UUID
@@ -315,7 +315,7 @@ def authenticate_credentials(
 
             if (
                 device_model.expires
-                and datetime.utcnow() >= device_model.expires
+                and datetime.now(timezone.utc) >= device_model.expires
             ):
                 error = (
                     f"Authentication error: device {decoded_token.device_id} "
@@ -552,7 +552,10 @@ def authenticate_device(client_id: UUID, device_code: str) -> AuthContext:
             error_description=error,
         )
 
-    if device_model.expires and datetime.utcnow() >= device_model.expires:
+    if (
+        device_model.expires
+        and datetime.now(timezone.utc) >= device_model.expires
+    ):
         error = (
             f"Authentication error: device for client ID {client_id} has "
             "expired"
@@ -835,17 +838,21 @@ def generate_access_token(
     if expires_in == 0:
         expires_in = None
     elif expires_in is not None:
-        expires = datetime.utcnow() + timedelta(seconds=expires_in)
+        expires = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
     elif device:
         # If a device was used for authentication, the token will expire
         # at the same time as the device.
         expires = device.expires
         if expires:
             expires_in = max(
-                int(expires.timestamp() - datetime.utcnow().timestamp()), 0
+                int(
+                    expires.timestamp()
+                    - datetime.now(timezone.utc).timestamp()
+                ),
+                0,
             )
     elif config.jwt_token_expire_minutes:
-        expires = datetime.utcnow() + timedelta(
+        expires = datetime.now(timezone.utc) + timedelta(
             minutes=config.jwt_token_expire_minutes
         )
         expires_in = config.jwt_token_expire_minutes * 60

src/zenml/zen_server/routers/devices_endpoints.py

@@ -13,7 +13,7 @@
 #  permissions and limitations under the License.
 """Endpoint definitions for code repositories."""
 
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Optional
 from uuid import UUID
 
@@ -219,7 +219,9 @@ def verify_authorized_device(
             )
 
         # Check if the device verification has expired.
-        if device_model.expires and device_model.expires < datetime.utcnow():
+        if device_model.expires and device_model.expires < datetime.now(
+            timezone.utc
+        ):
             raise ValueError(
                 "Invalid request: device verification expired.",
             )