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)

    29

    This website collects cookies to deliver better user experience

    Making a Youtube Browser with Python & React