Making a Youtube Browser with Python & React

Backend

In this series of articles, I'll talk about how to create a youtube browser (Basic) using Python with Flask for backend and React in the frontend, I'll divide all this into many parts

Backend (You are here 📍)

Frontend

Deploy

I created this Youtube Browser as a Technical Challenge for a role in my current job. It was so funny because I wanted to share my experience with you. Maybe I'll complete it by adding Unit Tests, CI/CD, and some features to keep it secure.

Well, here we go with the first part of this "mini-project". Let's create the backend!

In this case, I use one API with Python (Flask) this is built using Docker for this we make a file called Dockerfile with the code below:

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" ]

Now we'll create a file called requirements.txt, this file contains all the dependencies used in this microservice.

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

Now we need to create a docker-compose file to start the server on the local environment.

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}

As you can see, in the docker-compose file we have many environment variables one of these is ${API_KEY_CREDENTIALS} this variable is in a .env file and refer to API provided by Google to connect with this API, you can get this on your Google Console Dashboard.

Well now start with the real code, that was just the process to set up the container. We will create a folder called "app" that contains all the files used in this project.

Now create a file called app.py in this one we're going to add the code shown below to consult the Youtube API and return a response to the frontend.

First, we need to import all packages used in this file, some of them will be used in the future. For now add this and make the "check" route work ok.

# 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

Second, we will initialize Flask and set the logger level, additionally, we'll initialize the response manager.

# app.py

# Init Flask
app = Flask(__name__)

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

As you can see we have the package response_manager that contains a Class and this helps us (as the name reveals) to manage all the responses we make to the frontend. Let's create it and later instance it in the app.py file.

# 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')

Now we need an instance of the ResponseManager previously created, add the next code to do it.

# app.py

# Init ResponseManager
response_manager = ResponseManager(app)

Now add a route to check if the service is running ok and everything is fine, then add the next code.

# app.py

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

Next, we need to add the final code to make flask RUN, this make run on port 5000 and with the environment variable DEBUG we enable the debug on it.

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

At this point the app.py file looks like this.

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)

Ok now you can test it by making a request using Postman and verify if the container is working, if this is right you can go to the second step and add the route to request the Youtube API, and response to the frontend but first we need to create some utils functions and a YoutubeManager to help us do a "better" code, let's go with the 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')

So we have our YoutubeManager now let's go and create all utils functions, these functions will help us prepare the response to the frontend, I made it separate to isolate each part of code on small pieces for an easy debug in case of any problem.

# 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')

Well with this complete we can create finally our final function/route to request the Youtube API, add the code below in the 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)

Wait! But what will happen if someone sends bad parameters in the request?

Well, to solve it we'll add some query validations and respond to the message indicating something is wrong.

# 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)

Also created the file "models/schema.py" with the code below:

# 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()

Now add the next code @use_args(SearchSchema(), location='query') in the app.py file and the function will look like this.

# app.py

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

At the end, the app.py will looks like this:

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)