---
title: "Job Functionの設定"
url: https://docs.catalyst.zoho.com/ja/tutorials/newsapp/python/configure-job-function/
description: "Catalyst Job Scheduling Cronコンポーネントと、PythonでコーディングされたJob FunctionおよびAdvanced I/O関数を使用して、サードパーティサービスから定期的にニュースを自動取得するWebニュースアプリケーションを構築します。"
last_updated: 2026-08-17T12:16:04Z
service: All Services
type: tutorial
related_links:
  - name: "Job Scheduling"
    url: /ja/job-scheduling/getting-started/introduction/
  - name: "Job Functions"
    url: /ja/serverless/help/functions/job-functions/
---

# Job Functionの設定

次に、Job Functionコンポーネントを設定して、ニュースアプリケーションのコーディングを開始しましょう。

ファンクションのディレクトリ（この場合は`functions/NewsApp`）には、以下が含まれています:

* `main.py`メインファンクションファイル
* `catalyst-config.json`設定ファイル
* [requirements.txt](https://docs.catalyst.zoho.com/ja/cli/v1/project-directory-structure/functions-directory/#python-modules-and-function-dependencies)ファイル

`main.py`ファイルにコードを追加します。設定ファイルと依存関係ファイルのコードは変更しません。

[はじめに](https://docs.catalyst.zoho.com/ja/tutorials/newsapp/python/introduction/)で説明したように、Job Functionは2つのタスクを実行します: [NewsAPI](https://newsapi.org/)へのAPIコールを行いニュースを取得すること、およびCatalyst Data Storeにニュースを格納することです。APIコールにはNewsAPIが提供するAPIキーが使用されます。

### NewsAPIへの登録

Job Functionのコーディングを行う前に、NewsAPIの無料開発者サブスクリプションに登録し、以下の方法でAPIキーを取得する必要があります:

1. [https<span></span>://newsapi.org/register](https://newsapi.org/register)にアクセスします。

2. 必要な情報を入力し、**Submit**をクリックします。
![register-api-key](https://catalyst-docs-development.zohostratus.in/images/tutorials/cron/register-api-key.webp)
登録が完了すると、NewsAPIからAPIキーが提供されます。コードセクションの後の指示に従って、このキーをCronファンクションで使用する必要があります。
![api-key](https://catalyst-docs-development.zohostratus.in/images/tutorials/cron/api-key.webp)

### ファンクションコードの追加

以下のコードをコピーし、プロジェクトの`functions/NewsApp`ディレクトリにある`main.py`に貼り付けてファイルを保存します。アプリケーションのファイルの作業には任意のIDEを使用できます。

> **注意**: このセクションに記載されているコードを十分に理解してください。

**main.py**

```python
import logging
import zcatalyst_sdk
import requests
logger = logging.getLogger()
def handler(job_request, context):
    logger.info("Fetching news!")
    # ニュース項目を格納するData Store内のテーブル名
    table_name = ["HEADLINES", "BUSINESS", "ENTERTAINMENT", "HEALTH", "SCIENCE", "SPORTS", "TECHNOLOGY"]
    country = "US" # アメリカ合衆国からニュース項目を取得します
    api_key = "cc30ddcaxxxxxxxxxx" # NewsAPIから取得したAPIキーを引用符内に入力してください
    fetch_url = "https://newsapi.org/v2/top-headlines"
    # さまざまなカテゴリの速報ニュースをヘッドラインとともに取得します
    for category in table_name:
        # ニュース取得に必要なパラメータの構築
        params = {
            'country':country,
            'apiKey':api_key,
        }
        if category != "HEADLINES":
            params['category'] = category
        # News APIにAPIコールを行いデータを取得します
        response = requests.get(fetch_url, params=params)
        # レスポンスが200の場合、データはData Storeに移動されます。そうでない場合、エラーがログに記録されます。
        if response.status_code == 200:
            data = response.json()
            try:
                if 'articles' in data:
                    array_data = data['articles'][:15]
                    logger.info("15 days "+str(array_data))
                    # このメソッドはData Storeにニュースを挿入/更新します
                    push_to_datastore(array_data,category)
            except requests.RequestException as e:
                logger.error("Request failed: "+ str(e))
                context.close_with_failure() # アプリケーションを失敗で終了
            # アクションがログに記録されます。Catalyst Logsからログを確認できます。
            logger.info("New updated")
        else:
            logger.error("Request failed with status :: ",response.status_code)
            context.close_with_failure() # アプリケーションを失敗で終了
            context.close_with_success() # アプリケーションを成功で終了
def push_to_datastore(sliced_data,tableName):
    app = zcatalyst_sdk.initialize()
    zcql_service = app.zcql()
    datastore_service = app.datastore()
    # テーブル内の行数を確認するためのZCQLクエリを定義します
    query = "select ROWID from "+tableName
    zcql_result = zcql_service.execute_query(query)
    table_instance = datastore_service.table(tableName)
    # APIから取得したデータをData Storeに挿入/更新します
    for i in range(0,14):
        title = sliced_data[i]['title']
        url = sliced_data[i]['url']
        if len(zcql_result) == 0:
            # 新しい行を挿入
            row_data= {'title':title,'url':url}
            table_instance.insert_row(row_data)
            logger.info("Row inserted "+ str(row_data))
        else:
            # 行を更新
            row_data= {'title':title,'url':url,'ROWID':zcql_result[i][tableName]['ROWID']}
            table_instance.update_row(row_data)
            logger.info("Row inserted "+ str(row_data))

```

> **注意**: このコードをファンクションファイルにコピーして貼り付けた後、**12**行目の`APIKEY`の値を、NewsAPIから取得したAPIキーに置き換えてください。

Job Functionの設定が完了しました。クライアントの設定後に、アプリケーションのアーキテクチャについて説明します。