flask-kafka/basic_client.py - master - LarsFillmore/python_basic-flask-kafka

flask-kafka/basic_client.py
98 строк3 KB
SoullessSoldier
worked version 0.0.1
05 сен 2024, 20:44
05 сен 2024, 20:447505cd4
from typing import List, Dict, Union
from uuid import uuid4
import requests
from faker import Faker
import random
from datetime import datetime, timezone
import time

fake = Faker('ru_RU')

# URL = 'http://localhost:5000/submit'
URL = 'http://localhost:5000/post_data'
DATA_RANGE = 100


def get_users() -> List[Dict[str, str]]:
    """
    Generates 100 faked usernames.
    :return:
    List of 100 strings
    """
    usernames = []
    result_users = []
    while len(usernames) < 100:
        fake_name = fake.name()
        if fake.name not in usernames:
            usernames.append(fake_name)
            user = {'user_id': str(uuid4()),
                    'user_name': fake_name
                }
            result_users.append(user)

    return result_users


def get_films_stats() -> List[Dict[str, Union[str, int]]]:
    """
    Generates 100 film stat dictionaries.
    :return:
    List of 100 dicts
    """
    cat_titles = ['Рядовой', 'Сержант', 'Лейтенант', 'Капитан', 'Майор',
                  'Подполковник', 'Полковник', 'Генерал',
                  'Старшина', 'Детектив']
    cat_names = ['Пушистик', 'Пирожок', 'Котовски', 'Снежок', 'Пушок',
                 'Многошерст', 'Сосискин', 'Кусь', 'Розовые лапки', 'Мяуфрр']

    result = []
    result_list = []
    while len(result) < 100:
        name = f'{random.choice(cat_titles)} {random.choice(cat_names)}'
        if name not in result:
            result.append(name)
            result_dict = {
                           'film_title': name,
                           'film_duration': random.randrange(5, 160),
                           'film_genre': fake.color_name(),
                           'film_id': str(uuid4())
                           }
            result_list.append(result_dict)
    return result_list


def make_request(user_id: str, user_name: str, film_id: str, film_title: str,
                 film_duration: int, film_genre: str) -> int:
    date = datetime.now().replace(tzinfo=timezone.utc).isoformat()
    body = {
        'user_id': user_id, 'user_name': user_name, 'film_id': film_id,
        'film_title': film_title, 'film_duration': film_duration,
        'film_genre': film_genre,
        'timestamp': random.randrange(0, film_duration), 'date': date
    }
    response = requests.post(URL, json=body)
    return response.status_code


if __name__ == '__main__':
    users = get_users()
    films = get_films_stats()
    start = time.time()
    responses = {}
    for i in range(DATA_RANGE):
        user = random.choice(users)
        film = random.choice(films)
        print(user)
        response_code = make_request(user['user_id'], user['user_name'],
                                     film['film_id'], film['film_title'],
                                     film['film_duration'], film['film_genre'])
        if response_code in responses:
            responses[response_code] += 1
        else:
            responses[response_code] = 1

    end = time.time()
    print('Printing response stats:')
    for k, v in responses.items():
        print(f'{k}: {v}')
    print(f'Finished in {end - start} seconds')