12  Build A Django Content Aggregator

13 Django 팟캐스트 콘텐츠 수집기

참고: 이 프로젝트는 Python 3.9.1로 빌드되었지만 3.6 이상의 모든 Python 버전에서 작동해야 합니다.

13.1 이 저장소 정보

이는 _Real Python_에 대한 Build a Content Aggregator in Python 튜토리얼의 동반 프로젝트입니다. 기사를 방문하여 따라가거나 이 저장소에서 source_code_final/ 폴더의 콘텐츠를 다운로드하세요.

13.2 프로젝트 실행 방법

운영 체제에 맞는 Python 가상 환경을 생성 및 활성화하고 종속성을 설치합니다.

(.venv) $ python -m pip install -r <path_to_requirements.txt>

source_code_setup/requirements.txt에서 requirements.txt 파일을 찾을 수 있습니다.

source_code_final/으로 이동하여 Django 개발 서버를 시작합니다.

(.venv) $ cd source_code_final
(.venv) $ python manage.py runserver

이제 브라우저에서 localhost:8000으로 이동하여 완성된 프로젝트를 검사할 수 있습니다. 관련 튜토리얼에 설명된 대로 특정 단계에서 프로젝트를 살펴볼 수도 있습니다.

13.3 파일: source_code_final/content_aggregator/__init__.py

13.4 파일: source_code_final/content_aggregator/asgi.py

content_aggregator 프로젝트에 대한 ASGI 구성입니다.

이는 application이라는 모듈 수준 변수로 호출 가능한 ASGI를 노출합니다.

이 파일에 대한 자세한 내용은 다음을 참조하세요. https://docs.djangoproject.com/en/3.1/howto/deployment/asgi/

"""
ASGI config for content_aggregator project.

It exposes the ASGI callable as a module-level variable named ``application``.

For more information on this file, see
https://docs.djangoproject.com/en/3.1/howto/deployment/asgi/
"""

import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "content_aggregator.settings")

application = get_asgi_application()

13.5 파일: source_code_final/content_aggregator/settings.py

content_aggregator 프로젝트에 대한 Django 설정입니다.

Django 3.1.2를 사용하여 ’django-admin startproject’에 의해 생성되었습니다.

이 파일에 대한 자세한 내용은 다음을 참조하세요. https://docs.djangoproject.com/en/3.1/topics/settings/

설정 및 해당 값의 전체 목록은 다음을 참조하세요. https://docs.djangoproject.com/en/3.1/ref/settings/

"""
Django settings for content_aggregator project.

Generated by 'django-admin startproject' using Django 3.1.2.

For more information on this file, see
https://docs.djangoproject.com/en/3.1/topics/settings/

For the full list of settings and their values, see
https://docs.djangoproject.com/en/3.1/ref/settings/
"""

from pathlib import Path

# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent


# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/3.1/howto/deployment/checklist/

# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = "5kk6a_2ynd)yk!+uipwv5j-&!wu3v1e-*$&^*^lc_%tv+5ob4q"

# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = True

ALLOWED_HOSTS = []


# Application definition

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    # My Apps
    "podcasts.apps.PodcastsConfig",
    # Third Party Apps
    "django_apscheduler",
]

MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    "django.middleware.common.CommonMiddleware",
    "django.middleware.csrf.CsrfViewMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.contrib.messages.middleware.MessageMiddleware",
    "django.middleware.clickjacking.XFrameOptionsMiddleware",
]

ROOT_URLCONF = "content_aggregator.urls"

TEMPLATES = [
    {
        "BACKEND": "django.template.backends.django.DjangoTemplates",
        "DIRS": [
            BASE_DIR / "templates",
        ],
        "APP_DIRS": True,
        "OPTIONS": {
            "context_processors": [
                "django.template.context_processors.debug",
                "django.template.context_processors.request",
                "django.contrib.auth.context_processors.auth",
                "django.contrib.messages.context_processors.messages",
            ],
        },
    },
]

WSGI_APPLICATION = "content_aggregator.wsgi.application"


# Database
# https://docs.djangoproject.com/en/3.1/ref/settings/#databases

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.sqlite3",
        "NAME": BASE_DIR / "db.sqlite3",
    }
}


# Password validation
# https://docs.djangoproject.com/en/3.1/ref/settings/#auth-password-validators

AUTH_PASSWORD_VALIDATORS = [
    {
        "NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator",
    },
    {
        "NAME": "django.contrib.auth.password_validation.MinimumLengthValidator",
    },
    {
        "NAME": "django.contrib.auth.password_validation.CommonPasswordValidator",
    },
    {
        "NAME": "django.contrib.auth.password_validation.NumericPasswordValidator",
    },
]


# Internationalization
# https://docs.djangoproject.com/en/3.1/topics/i18n/

LANGUAGE_CODE = "en-us"

TIME_ZONE = "UTC"

USE_I18N = True

USE_L10N = True

USE_TZ = True


# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/3.1/howto/static-files/

STATIC_URL = "/static/"
STATICFILES_DIRS = [
    BASE_DIR / "static",
]

# Logging
# https://docs.djangoproject.com/en/3.1/topics/logging/

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
        },
    },
    "root": {
        "handlers": ["console"],
        "level": "INFO",
    },
}

13.6 파일: source_code_final/content_aggregator/urls.py

content_aggregator URL 구성

‘urlpatterns’ 목록은 URL을 뷰로 라우팅합니다. 자세한 내용은 다음을 참조하세요. https://docs.djangoproject.com/en/3.1/topics/http/urls/ 예: 기능 보기 1. 가져오기 추가: my_app 가져오기 보기에서 2. urlpatterns에 URL을 추가합니다: path(’‘, views.home, name=’home’) 클래스 기반 보기 1. 가져오기 추가: from other_app.views import 홈 2. urlpatterns에 URL을 추가합니다: path(’‘, Home.as_view(), name=’home’) 다른 URLconf 포함 1. include() 함수를 가져옵니다. django.urls에서 include, path를 가져옵니다. 2. urlpatterns에 URL을 추가합니다: path(‘blog/’, include(‘blog.urls’))

"""content_aggregator URL Configuration

The `urlpatterns` list routes URLs to views. For more information please see:
    https://docs.djangoproject.com/en/3.1/topics/http/urls/
Examples:
Function views
    1. Add an import:  from my_app import views
    2. Add a URL to urlpatterns:  path('', views.home, name='home')
Class-based views
    1. Add an import:  from other_app.views import Home
    2. Add a URL to urlpatterns:  path('', Home.as_view(), name='home')
Including another URLconf
    1. Import the include() function: from django.urls import include, path
    2. Add a URL to urlpatterns:  path('blog/', include('blog.urls'))
"""

from django.contrib import admin
from django.urls import include, path

urlpatterns = [
    path("admin/", admin.site.urls),
    path("", include("podcasts.urls")),
]

13.7 파일: source_code_final/content_aggregator/wsgi.py

content_aggregator 프로젝트에 대한 WSGI 구성입니다.

application이라는 모듈 수준 변수로 호출 가능한 WSGI를 노출합니다.

이 파일에 대한 자세한 내용은 다음을 참조하세요. https://docs.djangoproject.com/en/3.1/howto/deployment/wsgi/

"""
WSGI config for content_aggregator project.

It exposes the WSGI callable as a module-level variable named ``application``.

For more information on this file, see
https://docs.djangoproject.com/en/3.1/howto/deployment/wsgi/
"""

import os

from django.core.wsgi import get_wsgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "content_aggregator.settings")

application = get_wsgi_application()

13.8 파일: source_code_final/manage.py

관리 작업을 위한 Django의 명령줄 유틸리티입니다.

#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""

import os
import sys


def main():
    """Run administrative tasks."""
    os.environ.setdefault(
        "DJANGO_SETTINGS_MODULE", "content_aggregator.settings"
    )
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    execute_from_command_line(sys.argv)


if __name__ == "__main__":
    main()

13.9 파일: source_code_final/podcasts/__init__.py

13.10 파일: source_code_final/podcasts/admin.py

from django.contrib import admin

from .models import Episode


@admin.register(Episode)
class EpisodeAdmin(admin.ModelAdmin):
    list_display = ("podcast_name", "title", "pub_date")

13.11 파일: source_code_final/podcasts/apps.py

from django.apps import AppConfig


class PodcastsConfig(AppConfig):
    default_auto_field = "django.db.models.AutoField"
    name = "podcasts"

13.12 파일: source_code_final/podcasts/management/commands/startjobs.py

# Standard Library
import logging

# Third Party
import feedparser
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from dateutil import parser

# Django
from django.conf import settings
from django.core.management.base import BaseCommand
from django_apscheduler.jobstores import DjangoJobStore
from django_apscheduler.models import DjangoJobExecution

# Models
from podcasts.models import Episode

logger = logging.getLogger(__name__)


def save_new_episodes(feed):
    """Saves new episodes to the database.

    Checks the episode GUID agaist the episodes currently stored in the
    database. If not found, then a new `Episode` is added to the database.

    Args:
        feed: requires a feedparser object
    """
    podcast_title = feed.channel.title
    podcast_image = feed.channel.image["href"]

    for item in feed.entries:
        if not Episode.objects.filter(guid=item.guid).exists():
            episode = Episode(
                title=item.title,
                description=item.description,
                pub_date=parser.parse(item.published),
                link=item.link,
                image=podcast_image,
                podcast_name=podcast_title,
                guid=item.guid,
            )
            episode.save()


def fetch_realpython_episodes():
    """Fetches new episodes from RSS for the Real Python Podcast."""
    _feed = feedparser.parse("https://realpython.com/podcasts/rpp/feed")
    save_new_episodes(_feed)


def fetch_talkpython_episodes():
    """Fetches new episodes from RSS for the Talk Python to Me Podcast."""
    _feed = feedparser.parse("https://talkpython.fm/episodes/rss")
    save_new_episodes(_feed)


def delete_old_job_executions(max_age=604_800):
    """Deletes all apscheduler job execution logs older than `max_age`."""
    DjangoJobExecution.objects.delete_old_job_executions(max_age)


class Command(BaseCommand):
    help = "Runs apscheduler."

    def handle(self, *args, **options):
        scheduler = BlockingScheduler(timezone=settings.TIME_ZONE)
        scheduler.add_jobstore(DjangoJobStore(), "default")

        scheduler.add_job(
            fetch_realpython_episodes,
            trigger="interval",
            minutes=2,
            id="The Real Python Podcast",  # Each job MUST have a unique ID
            max_instances=1,
            # Replaces existing and stops duplicates on restart of the app.
            replace_existing=True,
        )
        logger.info("Added job: The Real Python Podcast.")

        scheduler.add_job(
            fetch_talkpython_episodes,
            trigger="interval",
            minutes=2,
            id="Talk Python Feed",
            max_instances=1,
            replace_existing=True,
        )
        logger.info("Added job: Talk Python Feed.")

        scheduler.add_job(
            delete_old_job_executions,
            trigger=CronTrigger(
                day_of_week="mon", hour="00", minute="00"
            ),  # Midnight on Monday, before start of the next work week.
            id="Delete Old Job Executions",
            max_instances=1,
            replace_existing=True,
        )
        logger.info("Added weekly job: Delete Old Job Executions.")

        try:
            logger.info("Starting scheduler...")
            scheduler.start()
        except KeyboardInterrupt:
            logger.info("Stopping scheduler...")
            scheduler.shutdown()
            logger.info("Scheduler shut down successfully!")

13.13 파일: source_code_final/podcasts/migrations/0001_initial.py

# Generated by Django 3.2.6 on 2021-10-03 13:38

from django.db import migrations, models


class Migration(migrations.Migration):

    initial = True

    dependencies = [
    ]

    operations = [
        migrations.CreateModel(
            name='Episode',
            fields=[
                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
                ('title', models.CharField(max_length=200)),
                ('description', models.TextField()),
                ('pub_date', models.DateTimeField()),
                ('link', models.URLField()),
                ('image', models.URLField()),
                ('podcast_name', models.CharField(max_length=100)),
                ('guid', models.CharField(max_length=50)),
            ],
        ),
    ]

13.14 파일: source_code_final/podcasts/migrations/__init__.py

13.15 파일: source_code_final/podcasts/models.py

from django.db import models


class Episode(models.Model):
    title = models.CharField(max_length=200)
    description = models.TextField()
    pub_date = models.DateTimeField()
    link = models.URLField()
    image = models.URLField()
    podcast_name = models.CharField(max_length=100)
    guid = models.CharField(max_length=50)

    def __str__(self) -> str:
        return f"{self.podcast_name}: {self.title}"

13.16 파일: source_code_final/podcasts/tests.py

from django.test import TestCase
from django.urls.base import reverse
from django.utils import timezone

from .models import Episode


class PodCastsTests(TestCase):
    def setUp(self):
        self.episode = Episode.objects.create(
            title="My Awesome Podcast Episode",
            description="Look mom, I made it!",
            pub_date=timezone.now(),
            link="https://myawesomeshow.com",
            image="https://image.myawesomeshow.com",
            podcast_name="My Python Podcast",
            guid="de194720-7b4c-49e2-a05f-432436d3fetr",
        )

    def test_episode_content(self):
        self.assertEqual(self.episode.description, "Look mom, I made it!")
        self.assertEqual(self.episode.link, "https://myawesomeshow.com")
        self.assertEqual(
            self.episode.guid, "de194720-7b4c-49e2-a05f-432436d3fetr"
        )

    def test_episode_str_representation(self):
        self.assertEqual(
            str(self.episode), "My Python Podcast: My Awesome Podcast Episode"
        )

    def test_home_page_status_code(self):
        response = self.client.get("/")
        self.assertEqual(response.status_code, 200)

    def test_home_page_uses_correct_template(self):
        response = self.client.get(reverse("homepage"))
        self.assertTemplateUsed(response, "homepage.html")

    def test_homepage_list_contents(self):
        response = self.client.get(reverse("homepage"))
        self.assertContains(response, "My Awesome Podcast Episode")

13.17 파일: source_code_final/podcasts/urls.py

from django.urls import path

from .views import HomePageView

urlpatterns = [
    path("", HomePageView.as_view(), name="homepage"),
]