blacklight/platypush
1
0
Fork 0
forked from platypush/platypush
Code Issues Pull requests Projects Releases Packages Wiki Activity

🐛 Partial revert of c18768e61f

Browse source 
`datetime.utcnow` may be deprecated on Python >= 3.12, but
`datetime.UTC` isn't present on older Python versions.

Added a `platypush.utils.utcnow()` method as a workaround compatible
with both.
This commit is contained in:
Fabio Manganiello 2024-05-31 19:52:32 +02:00
parent 4513bb9569
commit 3583dafbc3
Signed by: blacklight
GPG key ID: D90FBA7F76362774
12 changed files with 58 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
platypush/backend/http/app/routes/login.py
View file

@ -6,6 +6,7 @@ from flask import Blueprint, request, redirect, render_template, make_response
from platypush.backend.http.app import template_folder
from platypush.backend.http.utils import HttpUtils
from platypush.user import UserManager
from platypush.utils import utcnow
login = Blueprint('login', __name__, template_folder=template_folder)
@ -37,11 +38,7 @@ def login():
username = request.form.get('username')
password = request.form.get('password')
remember = request.form.get('remember')
expires = (
datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=365)
if remember
else None
)
expires = utcnow() + datetime.timedelta(days=365) if remember else None
session = user_manager.create_user_session(
username=username, password=password, expires_at=expires

5
platypush/backend/http/app/routes/register.py
View file

@ -6,6 +6,7 @@ from flask import Blueprint, request, redirect, render_template, make_response,
from platypush.backend.http.app import template_folder
from platypush.backend.http.utils import HttpUtils
from platypush.user import UserManager
from platypush.utils import utcnow
register = Blueprint('register', __name__, template_folder=template_folder)
@ -50,9 +51,7 @@ def register():
username=username,
password=password,
expires_at=(
datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=1)
if not remember
else None
utcnow() + datetime.timedelta(days=1) if not remember else None
),
)

8
platypush/entities/__init__.py
View file

@ -1,8 +1,10 @@
from datetime import datetime, timedelta, UTC
from datetime import timedelta
import logging
from threading import Event
from typing import Collection, Optional
from platypush.utils import utcnow
from ._base import (
Entity,
EntityKey,
@ -45,8 +47,8 @@ def get_entities_engine(timeout: Optional[float] = None) -> EntitiesEngine:
:param timeout: Timeout in seconds (default: None).
"""
time_start = datetime.now(UTC)
while not timeout or (datetime.now(UTC) - time_start < timedelta(seconds=timeout)):
time_start = utcnow()
while not timeout or (utcnow() - time_start < timedelta(seconds=timeout)):
if _engine:
break

11
platypush/entities/_base.py
View file

@ -6,7 +6,7 @@ import pathlib
import subprocess
import sys
import types
from datetime import datetime, UTC
from datetime import datetime
from typing import Callable, Dict, List, Optional, Set, Type, Tuple, Any
import pkgutil
@ -32,6 +32,7 @@ import platypush
from platypush.config import Config
from platypush.common.db import Base, is_defined
from platypush.message import JSONAble, Message
from platypush.utils import utcnow
EntityRegistryType = Dict[str, Type['Entity']]
entities_registry: EntityRegistryType = {}
@ -82,13 +83,11 @@ if not is_defined('entity'):
external_url = Column(String)
image_url = Column(String)
created_at = Column(
DateTime(timezone=False), default=datetime.now(UTC), nullable=False
)
created_at = Column(DateTime(timezone=False), default=utcnow(), nullable=False)
updated_at = Column(
DateTime(timezone=False),
default=datetime.now(UTC),
onupdate=datetime.now(UTC),
default=utcnow(),
onupdate=utcnow(),
)
parent = relationship(

5
platypush/entities/managers/__init__.py
View file

@ -1,12 +1,11 @@
import inspect
import json
from abc import ABC, abstractmethod
from datetime import datetime, UTC
from typing import Any, Optional, Dict, Collection, Type
from platypush.config import Config
from platypush.entities._base import Entity, EntitySavedCallback
from platypush.utils import get_plugin_name_by_class, get_redis
from platypush.utils import get_plugin_name_by_class, get_redis, utcnow
_entity_registry_varname = '_platypush/plugin_entity_registry'
@ -77,7 +76,7 @@ class EntityManager(ABC):
entity.id = None # type: ignore
entity.plugin = get_plugin_name_by_class(self.__class__) # type: ignore
entity.updated_at = datetime.now(UTC) # type: ignore
entity.updated_at = utcnow() # type: ignore
entity.children = self._normalize_entities(entity.children)
return entities

6
platypush/plugins/bluetooth/_ble/_event_handler.py
View file

@ -1,4 +1,4 @@
from datetime import datetime, timedelta, UTC
from datetime import timedelta
from logging import getLogger
from queue import Queue
from typing import Callable, Collection, Dict, Final, List, Optional, Type
@ -16,6 +16,7 @@ from platypush.message.event.bluetooth import (
BluetoothDeviceSignalUpdateEvent,
BluetoothDeviceEvent,
)
from platypush.utils import utcnow
from .._cache import EntityCache
from .._model import ServiceClass
@ -98,8 +99,7 @@ event_matchers: Dict[
)
and (
not (old and old.updated_at)
or datetime.now(UTC) - old.updated_at
>= timedelta(seconds=_rssi_update_interval)
or utcnow() - old.updated_at >= timedelta(seconds=_rssi_update_interval)
)
),
}

33
platypush/plugins/calendar/ical/__init__.py
View file

@ -4,6 +4,7 @@ from typing import Optional
from platypush.plugins import Plugin, action
from platypush.plugins.calendar import CalendarInterface
from platypush.utils import utcnow
class CalendarIcalPlugin(Plugin, CalendarInterface):
@ -25,7 +26,7 @@ class CalendarIcalPlugin(Plugin, CalendarInterface):
if not t:
return
if type(t.dt) == datetime.date:
if isinstance(t.dt, datetime.date):
return datetime.datetime(
t.dt.year, t.dt.month, t.dt.day, tzinfo=datetime.timezone.utc
).isoformat()
@ -42,21 +43,23 @@ class CalendarIcalPlugin(Plugin, CalendarInterface):
'id': str(event.get('uid')) if event.get('uid') else None,
'kind': 'calendar#event',
'summary': str(event.get('summary')) if event.get('summary') else None,
'description': str(event.get('description'))
if event.get('description')
else None,
'description': (
str(event.get('description')) if event.get('description') else None
),
'status': str(event.get('status')).lower() if event.get('status') else None,
'responseStatus': str(event.get('partstat')).lower()
if event.get('partstat')
else None,
'responseStatus': (
str(event.get('partstat')).lower() if event.get('partstat') else None
),
'location': str(event.get('location')) if event.get('location') else None,
'htmlLink': str(event.get('url')) if event.get('url') else None,
'organizer': {
'email': str(event.get('organizer')).replace('MAILTO:', ''),
'displayName': event.get('organizer').params.get('cn'),
}
if event.get('organizer')
else None,
'organizer': (
{
'email': str(event.get('organizer')).replace('MAILTO:', ''),
'displayName': event.get('organizer').params.get('cn'),
}
if event.get('organizer')
else None
),
'created': cls._convert_timestamp(event, 'created'),
'updated': cls._convert_timestamp(event, 'last-modified'),
'start': {
@ -95,9 +98,7 @@ class CalendarIcalPlugin(Plugin, CalendarInterface):
event['status'] != 'cancelled'
and event['end'].get('dateTime')
and event['end']['dateTime']
>= datetime.datetime.utcnow()
.replace(tzinfo=datetime.timezone.utc)
.isoformat()
>= utcnow().replace(tzinfo=datetime.timezone.utc).isoformat()
and (
(
only_participating

5
platypush/plugins/google/calendar/__init__.py
View file

@ -1,8 +1,7 @@
import datetime
from platypush.plugins import action
from platypush.plugins.google import GooglePlugin
from platypush.plugins.calendar import CalendarInterface
from platypush.utils import utcnow
class GoogleCalendarPlugin(GooglePlugin, CalendarInterface):
@ -72,7 +71,7 @@ class GoogleCalendarPlugin(GooglePlugin, CalendarInterface):
:meth:`platypush.plugins.calendar.CalendarPlugin.get_upcoming_events`.
"""
now = datetime.datetime.now(datetime.UTC).isoformat() + 'Z'
now = utcnow().isoformat() + 'Z'
service = self.get_service('calendar', 'v3')
result = (
service.events()

3
platypush/plugins/rss/__init__.py
View file

@ -16,6 +16,7 @@ from platypush.message.event.rss import NewFeedEntryEvent
from platypush.plugins import RunnablePlugin, action
from platypush.plugins.variable import VariablePlugin
from platypush.schemas.rss import RssFeedEntrySchema
from platypush.utils import utcnow
def _variable() -> VariablePlugin:
@ -289,7 +290,7 @@ class RssPlugin(RunnablePlugin):
title = ElementTree.Element('title')
title.text = 'Platypush feed subscriptions'
created = ElementTree.Element('dateCreated')
created.text = self._datetime_to_string(datetime.datetime.now(datetime.UTC))
created.text = self._datetime_to_string(utcnow())
head.append(title)
head.append(created)

3
platypush/plugins/sun/__init__.py
View file

@ -7,6 +7,7 @@ from dateutil.tz import gettz
from platypush.message.event.sun import SunriseEvent, SunsetEvent
from platypush.plugins import RunnablePlugin, action
from platypush.schemas.sun import SunEventsSchema
from platypush.utils import utcnow
class SunPlugin(RunnablePlugin):
@ -63,7 +64,7 @@ class SunPlugin(RunnablePlugin):
@staticmethod
def _convert_time(t: str) -> datetime.datetime:
now = datetime.datetime.now(datetime.UTC).replace(microsecond=0)
now = utcnow().replace(microsecond=0)
dt = datetime.datetime.strptime(
f'{now.year}-{now.month:02d}-{now.day:02d} {t}',
'%Y-%m-%d %I:%M:%S %p',

9
platypush/user/__init__.py
View file

@ -18,7 +18,7 @@ from platypush.exceptions.user import (
InvalidJWTTokenException,
InvalidCredentialsException,
)
from platypush.utils import get_or_generate_jwt_rsa_key_pair
from platypush.utils import get_or_generate_jwt_rsa_key_pair, utcnow
class UserManager:
@ -78,7 +78,7 @@ class UserManager:
),
password_salt=password_salt.hex(),
hmac_iterations=hmac_iterations,
created_at=datetime.datetime.now(datetime.UTC),
created_at=utcnow(),
**kwargs,
)
@ -116,8 +116,7 @@ class UserManager:
)
if not user_session or (
user_session.expires_at
and user_session.expires_at < datetime.datetime.now(datetime.UTC)
user_session.expires_at and user_session.expires_at < utcnow()
):
return None, None
@ -171,7 +170,7 @@ class UserManager:
user_id=user.user_id,
session_token=self.generate_session_token(),
csrf_token=self.generate_session_token(),
created_at=datetime.datetime.now(datetime.UTC),
created_at=utcnow(),
expires_at=expires_at,
)

12
platypush/utils/__init__.py
View file

@ -814,4 +814,16 @@ def wait_for_either(*events, timeout: Optional[float] = None, cls: Type = Event)
return OrEvent(*events, cls=cls).wait(timeout=timeout)
def utcnow():
"""
A workaround util to maintain compatibility both with Python >= 3.12 (which
deprecated datetime.utcnow) and Python < 3.12 (which doesn't have
datetime.UTC).
"""
if hasattr(datetime, 'UTC'):
return datetime.datetime.now(datetime.UTC)
return datetime.datetime.utcnow()
# vim:sw=4:ts=4:et: