Рубрики
Без рубрики

Сделать браузер YouTube с Python & React

Backend в этой серии статей, я расскажу о том, как создать браузер YouTube … Теги с Python, Docker, API.

Отдохнуть

В этой серии статей я расскажу о том, как создать браузер YouTube (Basic), используя Python с колбой для бэкэнда и реагировать на интерфейс, я делю все это во многих частях

  1. Backend (вы здесь 📍)
  2. Внешний интерфейс
  3. Развертывать

Я создал это Браузер YouTube как техническая задача для роли в моей нынешней работе. Это было так смешно, потому что я хотел поделиться своим опытом с вами. Может быть, я закончу это, добавив Установка тестов, CI/CD и некоторые функции, чтобы сохранить его в безопасности.

Ну, здесь мы идем с первой частью этого « Mini-Project ». Давайте создадим бэкэнду!

В этом случае я использую один API с Python (Колба) Это построено с помощью Docker для этого, мы сделаем файл под названием Dockerfile С кодом ниже:

FROM python:3.8-alpine

COPY requirements.txt .

RUN apk update

RUN pip install -r requirements.txt

COPY app /app

RUN ls -la

WORKDIR /app

EXPOSE 5000

ENTRYPOINT [ "./start.sh" ]

Теперь мы создадим файл под названием Требования .txt, этот файл содержит все зависимости, используемые в этом микросервисе.

Flask==2.0
gunicorn==20.1.0
pendulum==2.1.2
google-api-python-client
google-auth-oauthlib
google-auth-httplib2
oauth2client
webargs==8.0.0

Теперь нам нужно создать Docker-Compose Файл для запуска сервера в локальной среде.

version: "3.6"
services:
    youtube-search:
        container_name: youtube-search-service
        build: .
        image: python:latest
        ports:
            - 5000:5000
        volumes:
            - ./app/app.py:/app/app.py
            - ./app/utils.py:/app/utils.py
            - ./app/models/schema.py:/app/models/schema.py
            - ./app/youtube_manager.py:/app/youtube_manager.py
            - ./app/response_manager.py:/app/response_manager.py
        environment:
            TIMEOUT: 100
            DEBUG: "true"
            LOG_LEVEL: INFO
            TIMEZONE: America/Santiago
            DATE_FORMAT: YYYY-MM-DD HH:mm:ss
            API_KEY_CREDENTIALS: ${API_KEY_CREDENTIALS}

Как вы можете видеть, в Docker-Compose файле у нас много переменных среды Один из них – $ {Api_key_credentials} Эта переменная в .env Файл и обратитесь к API, предоставленным Google для подключения с этим API, вы можете получить это на вашем Приборная панель Google Console Отказ

Ну теперь начните с реального кода, это был просто процесс для настройки контейнера. Мы создадим папку «Приложение», которая содержит все файлы, используемые в этом проекте.

Теперь создайте файл под названием app.py В этом мы собираемся добавить код, показанный ниже, чтобы проконсультироваться с YouTube API и вернуть ответ на интерфейс.

Во-первых, нам нужно импортировать все пакеты, используемые в этом файле, некоторые из них будут использоваться в будущем. На данный момент добавьте это и сделайте «Проверка» Маршрут Работа в порядке.

# app.py

import json
import logging
from os import environ
from flask import Flask
from models.schema import SearchSchema
from webargs.flaskparser import use_args
from googleapiclient.errors import HttpError
from response_manager import ResponseManager
from youtube_manager import YoutubeManager
from marshmallow.exceptions import ValidationError
import utils as utils

Во-вторых, мы будем инициализировать колбу и установить уровень регистратора, дополнительно, мы инициализируем диспетчер ответов.

# app.py

# Init Flask
app = Flask(__name__)

# Set logger level
LOG_LEVEL = environ.get('LOG_LEVEL', 'INFO')
logging.basicConfig(level=LOG_LEVEL)

Как вы можете видеть, у нас есть пакет Ответ_manager который содержит Класс И это помогает нам (как название показывает) для управления всеми ответами, которые мы зарабатываем на интерфейс. Давайте создадим его и позже это в app.py файл.

# response_manager.py

import json
import logging
from os import environ

JSON_MIMETYPE='application/json'
LOG_LEVEL = environ.get('LOG_LEVEL', 'INFO')

# Set logger level
logging.basicConfig(level=LOG_LEVEL)

class ResponseManager():

    def __init__(self, app) -> None:
        self.app = app
        self.mimetype=JSON_MIMETYPE
        self.headers={'Access-Control-Allow-Origin': '*', 'Content-Type': JSON_MIMETYPE}
                # list of all possible errors
        self.list_status = { 400: 'fail', 422: 'fail', 500: 'error', 200:'success', 203: 'success' }
        self.status_code = None

    def message(self, message, status_code) -> dict:
        self.status_code = status_code
        try:
            return self.app.response_class(
                response = json.dumps({
                    'code': self.status_code
                    , 'status': self.list_status.get(self.status_code, 200) # default: 200
                    , 'message': message
                })
                , status=self.status_code
                , mimetype=self.mimetype
                , headers=self.headers
            )
        except Exception as error:
            logging.error(error)
            logging.exception('Something went wrong trying send message')

Теперь нам нужен экземпляр Ответник Ранее создан, добавьте следующий код, чтобы сделать это.

# app.py

# Init ResponseManager
response_manager = ResponseManager(app)

Теперь добавьте маршрут, чтобы проверить, работает ли услуга OK, и все в порядке, а затем добавьте следующий код.

# app.py

@app.route('/')
def index():
    return response_manager.message('Service is Working', 200)

Далее нам нужно добавить окончательный код, чтобы сделать Flask Run, это запускается в порту 5000 и с переменной окружающей среды ОТЛАЖИВАТЬ Мы включаем отладочную на нему.

if __name__ == '__main__':
    app.run(debug=environ.get('DEBUG', "true"), host='0.0.0.0', port=5000)

На данный момент app.py Файл выглядит так.

import json
import logging
from os import environ
from flask import Flask
from models.schema import SearchSchema
from webargs.flaskparser import use_args
from googleapiclient.errors import HttpError
from response_manager import ResponseManager
from youtube_manager import YoutubeManager
from marshmallow.exceptions import ValidationError
import utils as utils

# Init Flask
app = Flask(__name__)

# Set logger level
LOG_LEVEL = environ.get('LOG_LEVEL', 'INFO')
logging.basicConfig(level=LOG_LEVEL)

# Init ResponseManager
response_manager = ResponseManager(app)

@app.route('/')
def index():
    return response_manager.message('Service is Working', 200)

if __name__ == '__main__':
    app.run(debug=environ.get('DEBUG', "true"), host='0.0.0.0', port=5000)

Хорошо, теперь вы можете проверить его, сделав запрос, используя Почтальон И убедитесь, что контейнер работает, если это правильно, вы можете перейти на второй шаг и добавить маршрут, чтобы запросить YouTube API и ответ на интерфейс, но сначала нам нужно создать некоторые Utils Функции и A YouTubeManager Чтобы помочь нам сделать «лучше» код, давайте пойдем с YouTubeManager Отказ

# youtube_manager.py

import logging
from os import environ
from googleapiclient.discovery import build

# Set logger level
LOG_LEVEL = environ.get('LOG_LEVEL', 'INFO')
logging.basicConfig(level=LOG_LEVEL)

class YoutubeManager():

    def __init__(self, api_key) -> None:
        self.developer_api_key = api_key
        self.youtube_service_name = "youtube"
        self.youtube_api_version = "v3"
        self._youtube_cli = None

    def __enter__(self):
        try:
            logging.info('Initializing YouTube Manager')
            if self._youtube_cli is None:
                self._youtube_cli = build(
                    self.youtube_service_name
                    , self.youtube_api_version
                    , developerKey=self.developer_api_key
                )
            else:
                logging.info('Return existent client')
        except Exception as error:
            logging.error(error)
        else:
            logging.info('Returning YouTube Manager')
            return self._youtube_cli

    def __exit__(self, type, value, traceback) -> None:
        try:
            if self._youtube_cli is not None:
                self._youtube_cli = None

        except Exception as error:
            logging.error(error.args)
        else:
            logging.info('YouTube client closed')

Так что у нас есть наш YouTubeManager Теперь давайте пойдем и создам все функции Utils, эти функции помогут нам подготовить ответ на интерфейс, я сделал его отдельно для изоляции каждой части кода на небольших частях для легкой отладки в случае любой проблемы.

# utils.py

# Prepare the response 
def normalize_response(search_response, per_page = 100) -> dict:
    return {
        'prev_page': search_response.get('prevPageToken', None)
        , 'next_page': search_response.get('nextPageToken', None)
        , 'page_info': search_response.get('pageInfo', {'totalResults': 0, 'resultsPerPage': per_page})
        , 'items': normalize_items_response(search_response.get('items', []))
    }

# Prepare each item only with the required fields
def normalize_items_response(items) -> list:
    list_videos = []
    for item in items:
        item_id = item.get('id')
        item_snippet = item.get('snippet')
        item_thumbnails = item_snippet.get('thumbnails')

        new_item = {
            'id': get_id_item(item_id)
            , 'type': get_type_item(item_id.get('kind'))
            , 'description': item_snippet.get('description')
            , 'title': item_snippet.get('title')
            , 'channel_id': item_snippet.get('channelId')
            , 'channel_title': item_snippet.get('channelTitle')
            , 'published_at': item_snippet.get('publishedAt')
            , 'thumbnails': item_thumbnails.get('high')
        }
        list_videos.append(new_item)

    return list_videos

# Validate & Return the type of item
def get_type_item(kind):
    if kind == 'youtube#video':
        return 'video'
    elif kind == 'youtube#channel':
        return 'channel'
    else:
        return 'playlist'

# Validate & Return the ID according to each type
def get_id_item(item):
    if item.get('kind') == 'youtube#video':
        return item.get('videoId')
    elif item.get('kind') == 'youtube#channel':
        return item.get('channelId')
    else:
        return item.get('playlistId')

Хорошо с этим полным мы можем наконец создать нашу окончательную функцию/маршрут, чтобы запросить YouTube API Добавьте код ниже в app.py Отказ

# app.py

@app.route('/search', methods=['GET'])
def search_in_youtube(args):
    try:
        with YoutubeManager(environ.get('API_KEY_CREDENTIALS')) as youtube_manager:
            logging.info('Initializing search in Youtube API')
            max_results = args.get('per_page', 100)
            # Validating per_page parameter
            if (args.get('per_page') is not None):
                if int(args.get('per_page')) < 0:
                    raise ValidationError('Items per page must be greater than zero (0)')

            # We do the search using the YouTube API
            search_response = youtube_manager.search().list(
                q=args.get('query_search')
                , part='id, snippet'
                , maxResults=max_results
                , pageToken=args.get('page_token', None)
            ).execute()

            response = utils.normalize_response(search_response, max_results)

            return response_manager.message(response, 200)
    except ValidationError as error:
        logging.info(error)
        return response_manager.message(error.args, 422)
        # If is an HttpErrorException make send this message and log the error.
    except HttpError as error:
        error = json.loads(error.args[1])
        logging.error(error)
        return response_manager.message('Something went wrong searching in Youtube API', 500)
    except Exception as error:
        logging.error(error)
        return response_manager.message(error.args, 500)

Ждать! Но что произойдет, если кто-то отправит плохие параметры в запросе?

Ну, чтобы решить это, мы добавим некоторые проверки запроса и отвечайте на сообщение, указывающее, что что-то неверно.

# app.py

# Added function to send response when there is an error with status code 422
@app.errorhandler(422)
def validation_error(err):
    messages = err.data.get('messages')
    if 'query' in messages:
        return response_manager.message(messages.get('query'), 422)
    elif 'json' in messages:
        return response_manager.message(messages.get('json'), 422)

Также создал файл "Модели/схема .py" С кодом ниже:

# models/schema.py

from marshmallow import Schema, fields
from marshmallow.exceptions import ValidationError

class SearchSchema(Schema):
    per_page=fields.Integer()
    query_search=fields.String(required=True)
    page_token=fields.String()

Теперь добавьте следующий код @use_args (searchschema (),) в app.py Файл и функция будет выглядеть так.

# app.py

@app.route('/search', methods=['GET'])
@use_args(SearchSchema(), location='query') # -> Added code
def search_in_youtube(args):
    # search function code created before.

В конце концов, app.py будет выглядит так:

import json
import logging
from os import environ
from flask import Flask
from models.schema import SearchSchema
from webargs.flaskparser import use_args
from googleapiclient.errors import HttpError
from response_manager import ResponseManager
from youtube_manager import YoutubeManager
from marshmallow.exceptions import ValidationError
import utils as utils

# Init Flask
app = Flask(__name__)

# Set logger level
LOG_LEVEL = environ.get('LOG_LEVEL', 'INFO')
logging.basicConfig(level=LOG_LEVEL)

# Init ResponseManager
response_manager = ResponseManager(app)

# Added function to send response when there is an error with status code 422
@app.errorhandler(422)
def validation_error(err):
    messages = err.data.get('messages')
    if 'query' in messages:
        return response_manager.message(messages.get('query'), 422)
    elif 'json' in messages:
        return response_manager.message(messages.get('json'), 422)

@app.route('/')
def index():
    return response_manager.message('Service is Working', 200)

@app.route('/search', methods=['GET'])
@use_args(SearchSchema(), location='query')
def search_in_youtube(args):
    try:
        with YoutubeManager(environ.get('API_KEY_CREDENTIALS')) as youtube_manager:
            logging.info('Initializing search in Youtube API')
            max_results = args.get('per_page', 100)
            # Validating per_page parameter
            if (args.get('per_page') is not None):
                if int(args.get('per_page')) < 0:
                    raise ValidationError('Items per page must be greater than zero (0)')

            # We do the search using the YouTube API
            search_response = youtube_manager.search().list(
                q=args.get('query_search')
                , part='id, snippet'
                , maxResults=max_results
                , pageToken=args.get('page_token', None)
            ).execute()

            response = utils.normalize_response(search_response, max_results)

            return response_manager.message(response, 200)
    except ValidationError as error:
        logging.info(error)
        return response_manager.message(error.args, 422)
    except HttpError as error:
        error = json.loads(error.args[1])
        logging.error(error)
        return response_manager.message('Something went wrong searching in Youtube API', 500)
    except Exception as error:
        logging.error(error)
        return response_manager.message(error.args, 500)

if __name__ == '__main__':
    app.run(debug=environ.get('DEBUG', "true"), host='0.0.0.0', port=5000)

Оригинал: “https://dev.to/nestordgs/making-a-youtube-browser-with-python-react-4o1a”

Читайте ещё по теме: