Рубрика: Python

Здесь собраны статьи, наработки, исправление ошибок и решение проблем связанных с разработкой на языке Python

Python: Получение состояния реле Sonoff Basic R3

Python

Задача: получить состояние реле и записать его состояние в БД

Решение:

Реле отдает своё состояние по ссылке: 

http://{ip}:8081/zeroconf/info

Однако в зависимости от ревизии прошивки, данные могут немного отличаться. В нижеприведённом скрипте это учтено:

#!/usr/bin/python3
import config
import pymysql
import requests
from urllib3.exceptions import InsecureRequestWarning
import random

ip="192.168.88.245"             # IP реле
source=2        # источник - реле SonOff_1 (Гостинная)
place=1         # расположение реле - гостиннная

# здесь кусок кода который опрашивает текущее состояние реле
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
post_params = '{"deviceid": "","data": {}}'
response = requests.post(f"http://{ip}:8081/zeroconf/info", data=post_params, verify=False)
try:
    res = response.json()
    print(f"пришло:{res}")
    switch=True
    signal=0
    if type(res['data'])==str:
      res["data"]=json.loads(res["data"])
    if res["data"]["switch"]=="off":switch=False
    if "signalStrength" in res["data"]:signal=res["data"]["signalStrength"]

except Exception as e:
    print(f"Ошибка:{e}")
    exit(-1)

print(signal)
print(switch)

# соединяемся с БД
con=pymysql.connect(host=config.gconfig['mysql_host'],
                port=3306,
                user=config.gconfig['mysql_user'],
                password=config.gconfig['mysql_password'],
                database=config.gconfig['mysql_base'],
                cursorclass=pymysql.cursors.DictCursor
        )
# ложим данные в БД
with con.cursor() as cursor:
    sql=f"insert into m_data (place,source,value_type,value,dt) values ({place},{source},3,{switch},now())";
    cursor.execute(sql)
    con.commit()
    sql=f"insert into m_data (place,source,value_type,value,dt) values ({place},{source},4,{signal},now())";
    cursor.execute(sql)
    con.commit()
Оставить комментарий , , ,

Ротация логов в приложении на Python

Python

Задача: в приложении используется встроенный logging. Необходимо организовать ротацию логов со сжатием «архива»

Решение:

import coloredlogs,logging
from logging.handlers import TimedRotatingFileHandler
import gzip

class GZipRotator:
    def __call__(self, source, dest):
        os.rename(source, dest)
        f_in = open(dest, 'rb')
        f_out = gzip.open("%s.gz" % dest, 'wb')
        f_out.writelines(f_in)
        f_out.close()
        f_in.close()
        os.remove(dest)

rotation_logging_handler = TimedRotatingFileHandler(gconfig["logging_file"], when='D', interval=1, backupCount=5)
rotation_logging_handler.rotator=GZipRotator();
fh = logging.FileHandler('spam.log')
fh.filemode="a"

logger = logging.getLogger(__name__)
logging.basicConfig(format='%(asctime)s %(message)s',datefmt='%d.%m.%Y %H:%M:%S',level=gconfig["logging_level"],handlers=[rotation_logging_handler,fh])
logging.captureWarnings(True)

Результатом работы будет создание 1 раз в день сжатого файла

Оставить комментарий , , ,

Python: из объекта в JSON

Python

Если применять json.dumps с объектом, то можно словить ошибку сериализации объекта. Для обхода этого можно применить следующий способ:

def serialize(obj):    
    return obj.__dict__
loc=json.dumps(loc, default=serialize)

В этом случае, если штатный сериализатор «не справляется», то ему будет передана «подсказка» о типе

Оставить комментарий , ,

uvicorn: запускается только один инстанс при любом количестве workers

Python

Может поможет страждущим как и я. Двое суток гугла и янденкса ;(

Во время отладки скрипта на FastApi возникла ситуация, что сервис должен обращаться сам к себе, ну типа небольшая рекруссия. Соответсвенно чтоб трюк сработал, нужно чтобы было запущено несколько инстансов uvicorn одновременно. Ну ноу проблем, запускаю:

uvicorn main:app --reload --port 8000 --workers 10

И нифига. Запускается только один инстанс. И так и сяк, и гул и яндекс.

Наутро доходит попробовать убрать —reload (ну т.е. перезапускать инстанс автоматически при изменении кода — очень полезная штука во время отладки). И…..заработало! Баг?

Оставить комментарий

FastApi: передача заголовка Authorization в header

Python

Не тривиальная задача, которая потребовала двух дней поиска решения.

Казалось бы, всё решается просто:

@router.get('/versions',tags=["Credentials"],responses={
        200: {
            "model": List[models.versions_info],
            "description": "Return has code",
            "headers": {"Authorization": {"description":"Token party","type":"string"}}
        }})
async def list_versions(request: Request,token: Union[str, None] = Header(alias="Authorization",default=None)): 
    print(token)
    out=[{"version": "2.1.1","url": "https://www.server.com/ocpi/2.1.1/"},{"version": "2.2","url": "https://www.server.com/ocpi/2.2/"}]
    return Response(status_code=200,content=json.dumps(out), media_type="application/json", headers={"Authorization": "Token "+config.globals['mytoken']})

Однако вместо ожидаемой передачи заголовка, curl в упор возвращает ровно ничего:

Гугл помог выяснить, что ключевое слово Authorization зарезервированно для использования в Header. На https://stackoverflow.com было подсказано и направление куда копать. А именно, использовать обертку Security:

router = APIRouter()

token_key = fastapi.security.APIKeyHeader(name="Authorization")

class Token(BaseModel):
    token: str

def get_current_token(oauth_header: str = fastapi.Security(token_key)):
    print(f"Token: {oauth_header}")
    return oauth_header


# выдаем список поддерживаемых версий
@router.get('/versions',tags=["Credentials"],responses={
        200: {
            "model": List[models.versions_info],
            "description": "Return has code",
            "headers": {"Authorization": {"description":"Токен участника для авторизации","type":"string"}}
            },
        401: {
            "description": "Unauthorized status code.",
            }
        })
async def list_versions(request: Request,token: funcs.Token = Depends(funcs.get_current_token)): # запрос на получение поддерживаемых версий ПО
    print(f"Токен:{token}")
    if token==None: return Response(status_code=401, content="Unauthorized status code.", media_type="application/text")
    out=[{"version": "2.1.1","url": "https://www.server.com/ocpi/2.1.1/"},{"version": "2.2","url": "https://www.server.com/ocpi/2.2/"}]
    return Response(status_code=200,content=json.dumps(out), media_type="application/json", headers={"Authorization": "Token "+config.globals['mytoken']})
Оставить комментарий ,
1 … 3 4 5 6 7 … 13