Архивы python

Квест в консоли на Python. Часть 2

15.04.2024 Павел Грибов Python

Итак, продолжаем продолжаем писать квест в консоли на языке Python. Первая часть описана здесь. В ней мы реализовали автоматическую загрузку и сохранение состояния прохождения квеста. Сейчас же займемся (начнем по крайне мере) отрисовкой локации, и реакцией на нажатые кнопки. В локации предусмотрим возможность отображения картинки из ASCII. Например json стартовой локации может выглядеть примерно так:

{
  "title": "Стартовая локация",
  "description": "Стартовая локация описание",
  "ascii_art": "                    __            ================================\n         ALCATRAZ  /__\\            ||     ||<(.)>||<(.)>||     || \n       ____________|  |            ||    _||     ||     ||_    || \n       |_|_|_|_|_|_|  |            ||   (__D     ||     C__)   || \n       |_|_|_|_|_|_|__|            ||   (__D     ||     C__)   ||\n      A@\\|_|_|_|_|_|/@@Aa          ||   (__D     ||     C__)   ||\n   aaA@@@@@@@@@@@@@@@@@@@aaaA      ||   (__D     ||     C__)   ||\n  A@@@@@@@@@@@DWB@@@@@@@@@@@@A     ||     ||     ||     ||  dwb||\n^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^  ================================",
  "available_locations": {
    "left": 1,
    "right": 2,
    "forward": 3,
    "back": 4
  }
}

{

"title": "Стартовая локация",

"description": "Стартовая локация описание",

"ascii_art": " __ ================================\n ALCATRAZ /__\\ || ||<(.)>||<(.)>|| || \n ____________| | || _|| || ||_ || \n |_|_|_|_|_|_| | || (__D || C__) || \n |_|_|_|_|_|_|__| || (__D || C__) ||\n A@\\|_|_|_|_|_|/@@Aa || (__D || C__) ||\n aaA@@@@@@@@@@@@@@@@@@@aaaA || (__D || C__) ||\n A@@@@@@@@@@@DWB@@@@@@@@@@@@A || || || || dwb||\n^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ================================",

"available_locations": {

"left": 1,

"right": 2,

"forward": 3,

"back": 4

}

Создадим класс TLocation, при инициалиизации будем передавать в него инициализированный класс player. В переменной класса data — будем хранить загруженную локацию.

class TLocation:
    data = {}
    player={}
    stdscr=curses.initscr()
    scr_size=stdscr.getmaxyx()
    def __init__(self,player):
        self.player=player;
    def load_location(self, location):
        f = open("locations/" + str(location) + ".json", mode='r', encoding='utf-8')
        self.data = json.load(f)
        f.close()

class TLocation:

data = {}

player={}

stdscr=curses.initscr()

scr_size=stdscr.getmaxyx()

def __init__(self,player):

self.player=player;

def load_location(self, location):

f = open("locations/" + str(location) + ".json", mode='r', encoding='utf-8')

self.data = json.load(f)

f.close()

Далее нарисуем верхнее меню, где сообщаем игроку, на какие локации он может перемещаться, и что он держит в руках:

    def top_menu(self):
        loc="Идти: "
        if "left" in self.data["available_locations"]:
            loc=loc+"влево(4) "
        if "right" in self.data["available_locations"]:
            loc=loc+"вправо(6) "
        if "forward" in self.data["available_locations"]:
            loc=loc+"вперед(8) "
        if "back" in self.data["available_locations"]:
            loc=loc+"назад(2) "
        curses.start_color()
        curses.init_pair(1, curses.COLOR_RED, curses.COLOR_BLACK)
        self.stdscr.addstr(1, 1, loc,curses.color_pair(1))
        curses.init_pair(1, curses.COLOR_RED, curses.COLOR_BLACK)
        self.stdscr.addstr(1, 1, loc,curses.color_pair(1))
        # что в руках
        hands = "В руках: "
        self.stdscr.addstr(2, 1, "")
        for object in self.player.data["in_hands"]:
            hands=hands+"["+object+"]"
        if len(self.player.data["in_hands"])==0:
            hands = hands+"ничего нет"
        self.stdscr.addstr(2, 1, hands, curses.color_pair(1))

def top_menu(self):

loc="Идти: "

if "left" in self.data["available_locations"]:

loc=loc+"влево(4) "

if "right" in self.data["available_locations"]:

loc=loc+"вправо(6) "

if "forward" in self.data["available_locations"]:

loc=loc+"вперед(8) "

if "back" in self.data["available_locations"]:

loc=loc+"назад(2) "

curses.start_color()

curses.init_pair(1, curses.COLOR_RED, curses.COLOR_BLACK)

self.stdscr.addstr(1, 1, loc,curses.color_pair(1))

curses.init_pair(1, curses.COLOR_RED, curses.COLOR_BLACK)

self.stdscr.addstr(1, 1, loc,curses.color_pair(1))

# что в руках

hands = "В руках: "

self.stdscr.addstr(2, 1, "")

for object in self.player.data["in_hands"]:

hands=hands+"["+object+"]"

if len(self.player.data["in_hands"])==0:

hands = hands+"ничего нет"

self.stdscr.addstr(2, 1, hands, curses.color_pair(1))

Ну и собственно основной код отрисовки локации, включающий бесконечный цикл ожидания нажатий клавиатуры. Предусматриваем переход на другую локацию, выход из игры и сохранение игры.

    def location_view(self, location):
        self.load_location(location)
        self.stdscr.clear()
        self.stdscr.border()
        self.top_menu()  # рисуем верхнее меню
        # название локации
        curses.init_pair(2, curses.COLOR_GREEN, curses.COLOR_BLACK)
        x = int((self.scr_size[1] - len(self.data["title"])) / 2)
        self.stdscr.addstr(2, x, self.data["title"],curses.color_pair(2))
        # рисуем рисунок ежели он есть
        y=3
        if "ascii_art" in self.data:
            mass_art=self.data["ascii_art"].split("\n")
            i=0
            while i<len(mass_art):
                x=int((self.scr_size[1]-len(mass_art[i]))/2)
                self.stdscr.addstr(y+i, x, mass_art[i])
                i+=1
            y = y + len(mass_art);
         # Выводим описательную часть
        curses.init_pair(3, curses.COLOR_CYAN, curses.COLOR_BLACK)
        self.stdscr.addstr(y, 1, self.data["description"],curses.color_pair(3))
        # Выводим нижнее меню
        curses.init_pair(4, curses.COLOR_RED, curses.COLOR_BLACK)
        self.stdscr.addstr(y + 1, 1, "Осмотреться вокруг [v] Применить что в руках [h]", curses.color_pair(4))
        self.stdscr.addstr(y + 2, 1, "Выйти из квеста [q] Сохранить состояние [r]", curses.color_pair(4))
        self.stdscr.refresh()
        while True:
            key=self.stdscr.getch()
            print(key)
            # реализация перехода с локации на локацию
            if key==52 and "left" in self.data["available_locations"]:
                self.player.data["location"]=self.data["available_locations"]["left"]
                self.location_view(self.player.data["location"])
            if key==54 and "right" in self.data["available_locations"]:
                self.player.data["location"]=self.data["available_locations"]["right"]
                self.location_view(self.player.data["location"])
            if key==56 and "forward" in self.data["available_locations"]:
                self.player.data["location"]=self.data["available_locations"]["forward"]
                self.location_view(self.player.data["location"])
            if key==50 and "back" in self.data["available_locations"]:
                self.player.data["location"]=self.data["available_locations"]["back"]
                self.location_view(self.player.data["location"])
            if key == 114:
                self.player.save()
                #self.message("Внимание!","Состояние прохождения завершено. \nФайл находится в папке /saves")
            if key==113:
                curses.reset_shell_mode()
                curses.endwin()
                exit(0)
        print(self.data)

def location_view(self, location):

self.load_location(location)

self.stdscr.clear()

self.stdscr.border()

self.top_menu() # рисуем верхнее меню

# название локации

curses.init_pair(2, curses.COLOR_GREEN, curses.COLOR_BLACK)

x = int((self.scr_size[1] - len(self.data["title"])) / 2)

self.stdscr.addstr(2, x, self.data["title"],curses.color_pair(2))

# рисуем рисунок ежели он есть

y=3

if "ascii_art" in self.data:

mass_art=self.data["ascii_art"].split("\n")

i=0

while i<len(mass_art):

x=int((self.scr_size[1]-len(mass_art[i]))/2)

self.stdscr.addstr(y+i, x, mass_art[i])

i+=1

y = y + len(mass_art);

# Выводим описательную часть

curses.init_pair(3, curses.COLOR_CYAN, curses.COLOR_BLACK)

self.stdscr.addstr(y, 1, self.data["description"],curses.color_pair(3))

# Выводим нижнее меню

curses.init_pair(4, curses.COLOR_RED, curses.COLOR_BLACK)

self.stdscr.addstr(y + 1, 1, "Осмотреться вокруг [v] Применить что в руках [h]", curses.color_pair(4))

self.stdscr.addstr(y + 2, 1, "Выйти из квеста [q] Сохранить состояние [r]", curses.color_pair(4))

self.stdscr.refresh()

while True:

key=self.stdscr.getch()

print(key)

# реализация перехода с локации на локацию

if key==52 and "left" in self.data["available_locations"]:

self.player.data["location"]=self.data["available_locations"]["left"]

self.location_view(self.player.data["location"])

if key==54 and "right" in self.data["available_locations"]:

self.player.data["location"]=self.data["available_locations"]["right"]

self.location_view(self.player.data["location"])

if key==56 and "forward" in self.data["available_locations"]:

self.player.data["location"]=self.data["available_locations"]["forward"]

self.location_view(self.player.data["location"])

if key==50 and "back" in self.data["available_locations"]:

self.player.data["location"]=self.data["available_locations"]["back"]

self.location_view(self.player.data["location"])

if key == 114:

self.player.save()

#self.message("Внимание!","Состояние прохождения завершено. \nФайл находится в папке /saves")

if key==113:

curses.reset_shell_mode()

curses.endwin()

exit(0)

print(self.data)

В результате картинка (квест в консоли) на мониторе выглядит уже чуть симпатичнее:

Оставить комментарий python, движек, квест, консоль

Квест в консоли на Python. Часть 1

11.04.2024 Павел Грибов Python

Дело было вечером, делать было нечего (с). Ну не то чтобы совсем нечего, но выдалась свободное немножко время, поэтому для того чтобы не забыть (да уж чего там, и вспомнить уже) окончательно Python, решил сделать маленький движёк для текстовых квестов с выполняющихся в консоли (квест в консоли).

Сначала определимся что где и как:

Локации будем описывать в формате json
Локации будем складывать в папку locations. Имена файлов — номер локации.
В ходе квеста можно «сохраняться», чтобы была возможность продолжить квест
Сохранения будем хранить в папке saves
Все классы храним в папке classes

В результате у меня получилась такая структура папок и файлов:

Первым делом нарисую минимальный json стартовой локации:

{
  "title": "Стартовая страница",
  "description": "Стартовая страница описание",
  "available_locations": {
    "left": 1,
    "right": 2,
    "forward": 3,
    "back": 4
  }
}

{

"title": "Стартовая страница",

"description": "Стартовая страница описание",

"available_locations": {

"left": 1,

"right": 2,

"forward": 3,

"back": 4

}

Т.е. начинаем на стартовой локации (0), доступны переходы в локации 1,2,3 и 4.

Далее реализуем класс игрока, с реализацией функционала сохранения и стадии прохождения квеста:

from datetime import datetime as dt
import json
import os


class TPlayer:
    data = {}

    def __init__(self, name):
        self.data["name"] = name  # Имя пользователя
        self.data["location"] = 0  # текущая локация

    def load(self, filename):
        """
        Загрузить состояние квеста из файла
        :param filename: имя файла из папки saves
        """
        f = open("saves/" + filename, mode='r', encoding='utf-8')
        self.data = json.load(f)
        f.close()

    def list_saves(self):
        """
            Показать доступные сохранения
        """
        files = os.listdir("saves")
        print(files)

    def save(self):
        """
            Сохранить текущее состояние пользователя
        """
        time = dt.now()
        filename = time.strftime("%d_%m_%Y_%H_%M") + ".save"
        print(filename)
        f = open("saves/" + filename, mode='w', encoding='utf-8')
        json.dump(self.data, f)
        f.close()

from datetime import datetime as dt

import json

import os

class TPlayer:

data = {}

def __init__(self, name):

self.data["name"] = name # Имя пользователя

self.data["location"] = 0 # текущая локация

def load(self, filename):

"""

Загрузить состояние квеста из файла

:param filename: имя файла из папки saves

"""

f = open("saves/" + filename, mode='r', encoding='utf-8')

self.data = json.load(f)

f.close()

def list_saves(self):

"""

Показать доступные сохранения

"""

files = os.listdir("saves")

print(files)

def save(self):

"""

Сохранить текущее состояние пользователя

"""

time = dt.now()

filename = time.strftime("%d_%m_%Y_%H_%M") + ".save"

print(filename)

f = open("saves/" + filename, mode='w', encoding='utf-8')

json.dump(self.data, f)

f.close()

В главном файле (main.py), реализуем проверку аргументов командной строки и переход к началу квеста:

#!/usr/bin/env python3
# encoding: utf-8
import classes.player as tplayer
import classes.location as tlocation
import sys

player = tplayer.TPlayer("Васян")
location = tlocation.TLocation()


def start_location():
    print(player.data["name"])
    location.location_view(player.data["location"])

if __name__ == '__main__':
    for param in sys.argv:
        if param == "--load":
            if len(sys.argv) == 2:
                print("Ошибка: нет имени файла")
                exit(0)
            player.load(sys.argv[2])
            start_location()
            exit(0)
        if param == "--new":
            exit(0)
        if param == "--list":
            player.list_saves()
            exit(0)
    if len(sys.argv) == 1:
        print("Для запуска квеста необходимо использовать следующие параметры:")
        print("  --load   - загрузить сохранение и начать квест")
        print("  --list   - получить список сохранений")
        print("  --new <имя участника>    - начать квест заново")
        exit()

#!/usr/bin/env python3

# encoding: utf-8

import classes.player as tplayer

import classes.location as tlocation

import sys

player = tplayer.TPlayer("Васян")

location = tlocation.TLocation()

def start_location():

print(player.data["name"])

location.location_view(player.data["location"])

if __name__ == '__main__':

for param in sys.argv:

if param == "--load":

if len(sys.argv) == 2:

print("Ошибка: нет имени файла")

exit(0)

player.load(sys.argv[2])

start_location()

exit(0)

if param == "--new":

exit(0)

if param == "--list":

player.list_saves()

exit(0)

if len(sys.argv) == 1:

print("Для запуска квеста необходимо использовать следующие параметры:")

print(" --load - загрузить сохранение и начать квест")

print(" --list - получить список сохранений")

print(" --new <имя участника> - начать квест заново")

exit()

Разработка «квест в консоли» может быть действительно просто.. Вы можете посмотреть и другие мои статьи посвященные разработке на Python

Один комментарий python, квест в консоли

Сохранение весов модели нейросети

16.11.2023 Павел Грибов Python

В продолжении статьи Создаём нейросеть на Python, Итак сеть создали, натренировали. А что делать чтобы решение можно было определять по картинке лето или зима, не обучая каждый раз модель заново? Ну так ведь можно просто сохранить полученные веса нейросети в файл, А затем уже их загружать в случае необходимости.

В Tensotflow уже есть встроенный функционал для сохранения весов в файл. Для этого используется функция вызова кэлбека после прохождения каждого шага обучения. В моём случае промежуточные модели обучения не нужны, потому параметр save_freq делаю равным количеству итераций обучения:

traning_model_save=base_dir+"\save_model\cp.ckpt"
....
EPOCHS = 10
print("- настраиваем кэлбеки для сохранения натренированной модели")
# Создаем колбек для сохранения контрольной точки
cp_callback = tf.keras.callbacks.ModelCheckpoint(traning_model_save,
                                                 save_weights_only=True,
                                                 save_freq=EPOCHS,
                                                 verbose=1)
print("- тренируем модель")
history = model.fit(
    train_data_gen,
    steps_per_epoch=int(np.ceil(total_train / float(BATCH_SIZE))),
    epochs=EPOCHS,
    validation_data=val_data_gen,
    validation_steps=int(np.ceil(total_val / float(BATCH_SIZE))),
    callbacks = [cp_callback]   # вызываем кэлбек после каждого шага обучения
)

traning_model_save=base_dir+"\save_model\cp.ckpt"

....

EPOCHS = 10

print("- настраиваем кэлбеки для сохранения натренированной модели")

# Создаем колбек для сохранения контрольной точки

cp_callback = tf.keras.callbacks.ModelCheckpoint(traning_model_save,

save_weights_only=True,

save_freq=EPOCHS,

verbose=1)

print("- тренируем модель")

history = model.fit(

train_data_gen,

steps_per_epoch=int(np.ceil(total_train / float(BATCH_SIZE))),

epochs=EPOCHS,

validation_data=val_data_gen,

validation_steps=int(np.ceil(total_val / float(BATCH_SIZE))),

callbacks = [cp_callback] # вызываем кэлбек после каждого шага обучения

)

Так-же можно сохранять не только веса, но и всю модель целиком:

model.save('model.h5')

1	model.save('model.h5')

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

Пишем API правильно. На примере FastApi

08.09.2022 Павел Грибов Python

Долгое время был любителем по изобретать велосипед, и городил собственные реализации API для внешних систем. Да получалось. Да работало. Но если честно, всегда был в этих API бардак. Документации толковой не было, примеров соответственно тоже. Ну правда работал с этими API собственно практически один я, это и спасало от тухлых помидоров 😉

Решил попробовать таки в следующем своем проекте воспользоваться сторонними решениями для документирования и написания API. Выбор пал на FastApi для python. Собственно дальше просто «рыба» с основными фишками которую можно использовать далее расширяя.

Разберем файл main.py:

import uvicorn
from fastapi import FastAPI
from app.routers import hash

tags_metadata = [
    {"name": "use-hash","description": "Создание хэша по логину, паролю и секрету (если есть)"},
    {"name": "delete","description": "Удаление хэга по его  *ID*.",},
]

app = FastAPI(title="OCPI API documentation",description="Инструментарий для работы по протоколу OCPI",
    version="0.0.2",openapi_tags=tags_metadata,)

@app.on_event('startup')
async def startup():
    print("Приложение стартовало")

@app.on_event('shutdown')
async def shutdown():
    print("Приложение закрыто")

app.include_router(hash.router,prefix="/v1")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

import uvicorn

from fastapi import FastAPI

from app.routers import hash

tags_metadata = [

{"name": "use-hash","description": "Создание хэша по логину, паролю и секрету (если есть)"},

{"name": "delete","description": "Удаление хэга по его *ID*.",},

]

app = FastAPI(title="OCPI API documentation",description="Инструментарий для работы по протоколу OCPI",

version="0.0.2",openapi_tags=tags_metadata,)

@app.on_event('startup')

async def startup():

print("Приложение стартовало")

@app.on_event('shutdown')

async def shutdown():

print("Приложение закрыто")

app.include_router(hash.router,prefix="/v1")

if __name__ == "__main__":

uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

Что происходит в этом скрипте? Сначала инициализируется ядро FastApi:

app = FastAPI(title="OCPI API documentation",description="Инструментарий для работы по протоколу OCPI",
    version="0.0.2",openapi_tags=tags_metadata,)

app = FastAPI(title="OCPI API documentation",description="Инструментарий для работы по протоколу OCPI",

version="0.0.2",openapi_tags=tags_metadata,)

Т.е. обявляем название API, описание и присваиваем разделы которые будут отображаться в документации http://127.0.0.1:8000/docs

Далее подключаем роутинг из файла /app/routers/hash.py:

app.include_router(hash.router,prefix="/v1")

1	app.include_router(hash.router,prefix="/v1")

Этих роутингов естественно может быть не ограниченное количество

Для того чтобы FastApi мог сформировать документацию, необходимо её чуть подготовить. За это отвечают «схемы». Например создадим схему /app/schemas/hash.py:

# -*- coding: utf-8 -*-

from typing import Optional
from pydantic import BaseModel,Field

class hash_out(BaseModel):
    """ Что Выходит """
    hash: str = Field(..., title="Хэш созданый на основе логина пароля", example="iauwhfpseurh e")

class hash_in(BaseModel):
    """ Что входит """
    login: Optional [str] = Field(..., title="Логин пользователя", example="Например Вася")
    password: Optional [str]

class delete_param(BaseModel):
    id : Optional [int]

class delete_answer(BaseModel):
    result : Optional [str]

# -*- coding: utf-8 -*-

from typing import Optional

from pydantic import BaseModel,Field

class hash_out(BaseModel):

""" Что Выходит """

hash: str = Field(..., title="Хэш созданый на основе логина пароля", example="iauwhfpseurh e")

class hash_in(BaseModel):

""" Что входит """

password: Optional [str]

class delete_param(BaseModel):

id : Optional [int]

class delete_answer(BaseModel):

result : Optional [str]

Т.е. создаем классы с переменными которые могут быть входящими или исходящими пакетами работы API.

Далее создаем непосредственно файл работы с роутом hash:

# -*- coding: utf-8 -*-
"""Different helper-functions to work with users."""
import json

import fastapi
from fastapi import APIRouter,Depends,Request,Header,Response
from app.schemas import hash as models
import hashlib
from typing import Union

router = APIRouter()

@router.post('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры
async def hash(into : models.hash_in): # заданы входящие параметры
    # если параметры не заданы
    if into.password==None: into.password="NONE"
    if into.login == None: into.password = "NONE"
    hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()
    # сформируем ответ
    out=models.hash_out(hash=hash)
    return out

# -*- coding: utf-8 -*-

"""Different helper-functions to work with users."""

import json

import fastapi

from fastapi import APIRouter,Depends,Request,Header,Response

from app.schemas import hash as models

import hashlib

from typing import Union

router = APIRouter()

@router.post('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры

async def hash(into : models.hash_in): # заданы входящие параметры

# если параметры не заданы

if into.password==None: into.password="NONE"

if into.login == None: into.password = "NONE"

hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()

# сформируем ответ

out=models.hash_out(hash=hash)

return out

Что у нас тут? А вот что: мы объявляем, что FastAPI должен перенаправлять весь вход по методу POST с URL /hash (на самом деле /v1/hash, т.к. роутин объявили с префиксом v1) в функцию async def hash. В качестве исходящих значений, в response_model указываем схему hash_out заданную в схемах выше. Так -же установим тэг tags=[«use-hash»], для того чтоб функция API была красиво подписана. В качестве входящих параметров функции hash, указываем соответственно модель с классом hash_in

Если хочется «перехватывать» не только POST запросы, но и GET, то вызов чуть изменится:

@router.get('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры
async def hash(into : models.hash_in=Depends()): # заданы входящие параметры.

@router.get('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры

async def hash(into : models.hash_in=Depends()): # заданы входящие параметры.

Т.е. мы изменили вызов на get и в функцию добавили Depends(), т.к. входящие параметры попадают не в виде в данном случае JSON, а в качестве параметров вида: /v1/hash?login=aaa&passowd=bbb, т.е. их нужно конвертировать предварительно в JSON

А что делать, если помимо входящих параметров из POST запроса (ну или GET/PUT/DELETE и т.д), необходимо еще и обработать пришедшие заголовки Headers? Для этого изменим объявление функции, добавив помимо модели, еще и заголовки:

async def hash(into : models.hash_in=Depends(),secret: Union[str, None] = Header(description="Наш секрет",alias="Secret-Code-In",default=None)):

1	async def hash(into : models.hash_in=Depends(),secret: Union[str, None] = Header(description="Наш секрет",alias="Secret-Code-In",default=None)):

Тогда если в запросе будет присутствовать заголовок Secret-Code-In, то он будет передан в переменную secret

А если нам необходимо не только обработать входящие заголовки, но отдать их при ответе? Тут тогда получиться несколько сложнее.

@router.get('/hash_header',tags=["use-hash"],responses={
        200: {
            "model": models.hash_out,
            "description": "Return has code",
            "headers": {
                "Secret-Code-Out": {"description":"Secret code out","type":"string"}
            }
        }

@router.get('/hash_header',tags=["use-hash"],responses={

200: {

"model": models.hash_out,

"description": "Return has code",

"headers": {

"Secret-Code-Out": {"description":"Secret code out","type":"string"}

}

Т.е. нам придется использовать не обычно используемый response_model, для указания класса со схемой ответа, а response, который позволяет указать шаблон при определенном коде ответа. В результате общий файл роута hash.py получается такой:

# -*- coding: utf-8 -*-
"""Different helper-functions to work with users."""
import json

import fastapi
from fastapi import APIRouter,Depends,Request,Header,Response
from app.schemas import hash as models
import hashlib
from typing import Union

router = APIRouter()

@router.post('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры
async def hash(into : models.hash_in): # заданы входящие параметры
    # если параметры не заданы
    if into.password==None: into.password="NONE"
    if into.login == None: into.password = "NONE"
    hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()
    # сформируем ответ
    out=models.hash_out(hash=hash)
    return out

@router.get('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры
async def hash(into : models.hash_in=Depends()): # заданы входящие параметры.
    # Внимание! Обращаяю внимание на Depends(), оно избавляет от ошибки "Request with GET/HEAD method cannot have body",
    # т.е. так параметры ищутся не в body, а в самом запросе GET в виде параметров

    # если параметры не заданы
    if into.password==None: into.password="NONE"
    if into.login == None: into.password = "NONE"
    hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()
    # сформируем ответ
    out=models.hash_out(hash=hash)
    return out

@router.put('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры
async def hash(into : models.hash_in): # заданы входящие параметры
    # если параметры не заданы
    if into.password==None: into.password="NONE"
    if into.login == None: into.password = "NONE"
    hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()
    # сформируем ответ
    out=models.hash_out(hash=hash)
    return out

@router.delete('/hash',response_model=models.delete_answer,tags=["delete"]) # заданы исходящие параметры
async def hash(into : models.delete_param): # заданы входящие параметры
    # сформируем ответ
    out=models.delete_answer(result="Ok")
    return out

@router.get('/hash_header',tags=["use-hash"],responses={
        200: {
            "model": models.hash_out,
            "description": "Return has code",
            "headers": {
                "Secret-Code-Out": {"description":"Secret code out","type":"string"}
            }
        }
}) # заданы исходящие параметры
async def hash(into : models.hash_in=Depends(),secret: Union[str, None] = Header(description="Наш секрет",alias="Secret-Code-In",default=None)):
    hash = hashlib.sha1((into.login + into.password+secret).encode('utf-8')).hexdigest()
    # сформируем ответ
    out = models.hash_out(hash=hash)
    return Response(status_code=200,content=json.dumps(out.__dict__), media_type="application/json", headers={"Secret-Code-Out": secret})

# -*- coding: utf-8 -*-

"""Different helper-functions to work with users."""

import json

import fastapi

from fastapi import APIRouter,Depends,Request,Header,Response

from app.schemas import hash as models

import hashlib

from typing import Union

router = APIRouter()

@router.post('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры

async def hash(into : models.hash_in): # заданы входящие параметры

# если параметры не заданы

if into.password==None: into.password="NONE"

if into.login == None: into.password = "NONE"

hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()

# сформируем ответ

out=models.hash_out(hash=hash)

return out

@router.get('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры

async def hash(into : models.hash_in=Depends()): # заданы входящие параметры.

# Внимание! Обращаяю внимание на Depends(), оно избавляет от ошибки "Request with GET/HEAD method cannot have body",

# т.е. так параметры ищутся не в body, а в самом запросе GET в виде параметров

# если параметры не заданы

if into.password==None: into.password="NONE"

if into.login == None: into.password = "NONE"

hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()

# сформируем ответ

out=models.hash_out(hash=hash)

return out

@router.put('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры

async def hash(into : models.hash_in): # заданы входящие параметры

# если параметры не заданы

if into.password==None: into.password="NONE"

if into.login == None: into.password = "NONE"

hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()

# сформируем ответ

out=models.hash_out(hash=hash)

return out

@router.delete('/hash',response_model=models.delete_answer,tags=["delete"]) # заданы исходящие параметры

async def hash(into : models.delete_param): # заданы входящие параметры

# сформируем ответ

out=models.delete_answer(result="Ok")

return out

@router.get('/hash_header',tags=["use-hash"],responses={

200: {

"model": models.hash_out,

"description": "Return has code",

"headers": {

"Secret-Code-Out": {"description":"Secret code out","type":"string"}

}

}) # заданы исходящие параметры

async def hash(into : models.hash_in=Depends(),secret: Union[str, None] = Header(description="Наш секрет",alias="Secret-Code-In",default=None)):

hash = hashlib.sha1((into.login + into.password+secret).encode('utf-8')).hexdigest()

# сформируем ответ

out = models.hash_out(hash=hash)

return Response(status_code=200,content=json.dumps(out.__dict__), media_type="application/json", headers={"Secret-Code-Out": secret})

Что в принципе покрывает используемые в большинстве случаев кейсы использования

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

Оформление службы в systemd из файла python

05.09.2022 Павел Грибов Python

Для того чтобы служба полноценно работала, в файле python необходимо предусмотреть:

Блокировку запуска копии скрипта
Создание pid файла с номером процесса

Теоретически это возможно возложить и на плечи systemd, но «классически» делать это самому.

Пример создания pid файла:

import os
    # выясним id процесса т создадим pid файл
    try:
        pid=os.getpid()
        with open(f"/var/run/ocpp_{sys.argv[1]}.pid", "w") as file:
            file.write(str(pid))
    except Exception as e:
        functions.logapi.error(f"Не удалось создать PID файл {e}");

import os

# выясним id процесса т создадим pid файл

try:

pid=os.getpid()

with open(f"/var/run/ocpp_{sys.argv[1]}.pid", "w") as file:

file.write(str(pid))

except Exception as e:

functions.logapi.error(f"Не удалось создать PID файл {e}");

Пример блокировки запуска копии:

from filelock import FileLock,Timeout    
    lock = FileLock(f"{sys.argv[1]}.lock",0)
    try:
        lock.acquire()
    except Timeout:
        functions.logapi.error(f"инстанс {sys.argv[1]} уже запущен на сервере");
        exit(-1);
    with lock:
        functions.logapi.debug(f"включена блокировка запусков других инстансов");

from filelock import FileLock,Timeout

lock = FileLock(f"{sys.argv[1]}.lock",0)

try:

lock.acquire()

except Timeout:

functions.logapi.error(f"инстанс {sys.argv[1]} уже запущен на сервере");

exit(-1);

with lock:

functions.logapi.debug(f"включена блокировка запусков других инстансов");

Пример файла настройки службы в этом случае (xx.service):

[Unit]
Description=ocpp1

[Service]
ExecStart=python3 /home/user/цувцув/цувцу.py ocpp1
PIDFile=/var/run/ocpp_ocpp1.pid

[Install]
WantedBy=multi-user.target

[Unit]

Description=ocpp1

[Service]

ExecStart=python3 /home/user/цувцув/цувцу.py ocpp1

PIDFile=/var/run/ocpp_ocpp1.pid

[Install]

WantedBy=multi-user.target

В Ubuntu его необходимо положить в /etc/systemd/system и перезапустить службу:

sudo systemctl daemon-reload

1	sudo systemctl daemon-reload

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

1 2 3 … 8 »

Ноябрь 2024
Пн	Вт	Ср	Чт	Пт	Сб	Вс
				1	2	3
4	5	6	7	8	9	10
11	12	13	14	15	16	17
18	19	20	21	22	23	24
25	26	27	28	29	30