ວຽກງານພື້ນຖານກ່ຽວກັບ Faust, ພາກທີ II: ຕົວແທນ ແລະທີມງານ

ຕາຕະລາງເນື້ອຫາ

1. ພາກທີ I: ແນະນໍາ

2. ພາກທີ II: ຕົວແທນແລະທີມງານ

ພວກເຮົາເຮັດຫຍັງຢູ່ນີ້?

ດັ່ງນັ້ນ, ດັ່ງນັ້ນ, ສ່ວນທີສອງ. ດັ່ງທີ່ຂຽນໄວ້ກ່ອນຫນ້ານີ້, ໃນມັນພວກເຮົາຈະເຮັດດັ່ງຕໍ່ໄປນີ້:

1. ໃຫ້ຂຽນລູກຄ້າຂະຫນາດນ້ອຍສໍາລັບ alphavantage ໃນ aiohttp ດ້ວຍການຮ້ອງຂໍສໍາລັບຈຸດສິ້ນສຸດທີ່ພວກເຮົາຕ້ອງການ.

2. ໃຫ້ສ້າງຕົວແທນຜູ້ທີ່ຈະເກັບກໍາຂໍ້ມູນຫຼັກຊັບແລະຂໍ້ມູນ meta ກ່ຽວກັບພວກເຂົາ.

ແຕ່, ນີ້ແມ່ນສິ່ງທີ່ພວກເຮົາຈະເຮັດສໍາລັບໂຄງການຕົວມັນເອງ, ແລະໃນແງ່ຂອງການຄົ້ນຄວ້າ faust, ພວກເຮົາຈະຮຽນຮູ້ວິທີການຂຽນຕົວແທນທີ່ຂະບວນການຖ່າຍທອດເຫດການຈາກ kafka, ເຊັ່ນດຽວກັນກັບວິທີການຂຽນຄໍາສັ່ງ (click wrapper), ໃນກໍລະນີຂອງພວກເຮົາ - ສໍາລັບຂໍ້ຄວາມຊຸກຍູ້ຄູ່ມືໄປຫາຫົວຂໍ້ທີ່ຕົວແທນກໍາລັງຕິດຕາມ.

ການ​ຝຶກ​ອົບ​ຮົມ

ລູກຄ້າ AlphaVantage

ທໍາອິດ, ໃຫ້ຂຽນລູກຄ້າ aiohttp ຂະຫນາດນ້ອຍສໍາລັບການຮ້ອງຂໍໃຫ້ alphavantage.

alphavantage.py

Spoiler

import urllib.parse as urlparse
from io import StringIO
from typing import Any, Dict, List, Union

import aiohttp
import pandas as pd
import stringcase
from loguru import logger

from horton.config import API_ENDPOINT


class AlphaVantageClient:
    def __init__(
        self,
        session: aiohttp.ClientSession,
        api_key: str,
        api_endpoint: str = API_ENDPOINT,
    ):
        self._query_params = {"datatype": "json", "apikey": api_key}
        self._api_endpoint = api_endpoint
        self._session = session

    @logger.catch
    def _format_fields(self, data: Dict[str, Any]) -> Dict[str, Any]:
        formatted_data = {}

        for field, item in data.items():
            formatted_data[stringcase.snakecase(field)] = item

        return formatted_data

    @logger.catch
    async def _construct_query(
        self, function: str, to_json: bool = True, **kwargs
    ) -> Union[Dict[str, Any], str]:
        path = "query/"

        async with self._session.get(
            urlparse.urljoin(self._api_endpoint, path),
            params={"function": function, **kwargs, **self._query_params},
        ) as response:
            data = (await response.json()) if to_json else (await response.text())

            if to_json:
                data = self._format_fields(data)

        return data

    @logger.catch
    async def get_securities(self, state: str = "active") -> List[Dict[str, str]]:
        data = await self._construct_query("LISTING_STATUS", state=state, to_json=False)

        data = pd.read_csv(StringIO(data))

        securities = data.to_dict("records")

        for index, security in enumerate(securities):
            security = self._format_fields(security)
            security["_type"] = "physical"

            securities[index] = security

        return securities

    @logger.catch
    async def get_security_overview(self, symbol: str) -> Dict[str, str]:
        return await self._construct_query("OVERVIEW", symbol=symbol)

    @logger.catch
    async def get_historical_data(self, symbol: str) -> Dict[str, Any]:
        return await self._construct_query(
            "TIME_SERIES_DAILY_ADJUSTED", symbol=symbol, outputsize="full"
        )

    @logger.catch
    async def get_last_price_data(self, symbol: str) -> Dict[str, Any]:
        return await self._construct_query("GLOBAL_QUOTE", symbol=symbol)

    @logger.catch
    async def get_indicator_data(
        self, symbol: str, indicator: str, **indicator_options
    ) -> Dict[str, Any]:
        return await self._construct_query(
            indicator, symbol=symbol, **indicator_options
        )

ໃນຄວາມເປັນຈິງ, ທຸກສິ່ງທຸກຢ່າງແມ່ນຈະແຈ້ງຈາກມັນ:

1. AlphaVantage API ແມ່ນຂ້ອນຂ້າງງ່າຍດາຍແລະອອກແບບທີ່ສວຍງາມ, ດັ່ງນັ້ນຂ້ອຍຈຶ່ງຕັດສິນໃຈເຮັດການຮ້ອງຂໍທັງຫມົດໂດຍຜ່ານວິທີການ construct_query ບ່ອນທີ່ມີການໂທ http.

2. ຂ້ອຍເອົາທົ່ງນາທັງໝົດມາໃຫ້ snake_case ເພື່ອຄວາມສະດວກ.

3. ດີ, ການຕົບແຕ່ງ logger.catch ສໍາລັບຜົນຜະລິດ traceback ທີ່ສວຍງາມແລະໃຫ້ຂໍ້ມູນ.

PS ຢ່າລືມເພີ່ມ token alphavantage ຢູ່ໃນທ້ອງຖິ່ນເພື່ອ config.yml, ຫຼືສົ່ງອອກຕົວແປສະພາບແວດລ້ອມ HORTON_SERVICE_APIKEY. ພວກເຮົາໄດ້ຮັບ token ທີ່ນີ້.

ຫ້ອງຮຽນ CRUD

ພວກເຮົາຈະມີການເກັບກໍາຫຼັກຊັບເພື່ອເກັບຂໍ້ມູນ meta ກ່ຽວກັບຫຼັກຊັບ.

ຖານຂໍ້ມູນ/security.py

ໃນຄວາມຄິດເຫັນຂອງຂ້ອຍ, ບໍ່ຈໍາເປັນຕ້ອງອະທິບາຍຫຍັງຢູ່ທີ່ນີ້, ແລະຊັ້ນພື້ນຖານຂອງມັນເອງແມ່ນຂ້ອນຂ້າງງ່າຍດາຍ.

get_app()

ໃຫ້ເພີ່ມຟັງຊັນສ້າງວັດຖຸແອັບພລິເຄຊັນໃນ app.py

Spoiler

import faust

from horton.config import KAFKA_BROKERS


def get_app():
    return faust.App("horton", broker=KAFKA_BROKERS)

ສໍາລັບໃນປັດຈຸບັນພວກເຮົາຈະມີການສ້າງຄໍາຮ້ອງສະຫມັກທີ່ງ່າຍດາຍທີ່ສຸດ, ເລັກນ້ອຍຕໍ່ມາພວກເຮົາຈະຂະຫຍາຍມັນ, ຢ່າງໃດກໍຕາມ, ເພື່ອບໍ່ໃຫ້ທ່ານລໍຖ້າ, ທີ່ນີ້. ອ້າງອີງ ກັບ App-class. ຂ້າ​ພະ​ເຈົ້າ​ຍັງ​ແນະ​ນໍາ​ໃຫ້​ທ່ານ​ເບິ່ງ​ໃນ​ຫ້ອງ​ການ​ປັບ​ຄ່າ​, ເນື່ອງ​ຈາກ​ວ່າ​ມັນ​ມີ​ຄວາມ​ຮັບ​ຜິດ​ຊອບ​ສໍາ​ລັບ​ການ​ຕັ້ງ​ຄ່າ​ຫຼາຍ​ທີ່​ສຸດ​.

ສ່ວນໃຫຍ່ແມ່ນ

ຕົວແທນສໍາລັບການເກັບແລະຮັກສາບັນຊີລາຍຊື່ຂອງຫຼັກຊັບ

app = get_app()

collect_securities_topic = app.topic("collect_securities", internal=True)

@app.agent(collect_securities_topic)
async def collect_securities(stream: StreamT[None]) -> AsyncIterable[bool]:
	pass

ດັ່ງນັ້ນ, ທໍາອິດພວກເຮົາໄດ້ຮັບວັດຖຸຄໍາຮ້ອງສະຫມັກ faust - ມັນງ່າຍດາຍດີ. ຕໍ່ໄປ, ພວກເຮົາປະກາດຫົວຂໍ້ສໍາລັບຕົວແທນຂອງພວກເຮົາຢ່າງຈະແຈ້ງ ... ໃນທີ່ນີ້ມັນຄວນຈະເປັນການກ່າວເຖິງວ່າມັນແມ່ນຫຍັງ, ຕົວກໍານົດການພາຍໃນແມ່ນຫຍັງແລະວິທີການນີ້ສາມາດໄດ້ຮັບການຈັດລຽງແຕກຕ່າງກັນ.

1. ຫົວຂໍ້ໃນ kafka, ຖ້າພວກເຮົາຕ້ອງການຮູ້ຄໍານິຍາມທີ່ແນ່ນອນ, ມັນດີກວ່າທີ່ຈະອ່ານ ປິດ. ເອກະສານ, ຫຼືທ່ານສາມາດອ່ານ ບົດສະຫຼຸບ on Habre ໃນ​ພາ​ສາ​ລັດ​ເຊຍ​, ບ່ອນ​ທີ່​ທຸກ​ສິ່ງ​ທຸກ​ຢ່າງ​ແມ່ນ​ສະ​ທ້ອນ​ໃຫ້​ເຫັນ​ຢ່າງ​ຖືກ​ຕ້ອງ :)

2. ພາຣາມິເຕີພາຍໃນ, ອະທິບາຍໄດ້ດີໃນ faust doc, ອະນຸຍາດໃຫ້ພວກເຮົາກໍານົດຫົວຂໍ້ໂດຍກົງໃນລະຫັດ, ແນ່ນອນ, ນີ້ຫມາຍຄວາມວ່າຕົວກໍານົດການສະຫນອງໃຫ້ໂດຍນັກພັດທະນາ faust, ຕົວຢ່າງ: ການຮັກສາໄວ້, ນະໂຍບາຍການເກັບຮັກສາ (ໂດຍຄ່າເລີ່ມຕົ້ນລຶບ, ແຕ່ທ່ານສາມາດກໍານົດ. ຫນາແຫນ້ນ), ຈໍາ​ນວນ​ຂອງ​ການ​ແບ່ງ​ປັນ​ຕໍ່​ຫົວ​ຂໍ້ (ຄະແນນເພື່ອເຮັດ, ສໍາລັບການຍົກຕົວຢ່າງ, ຫນ້ອຍກວ່າ ຄວາມສໍາຄັນທົ່ວໂລກ faust ຄໍາຮ້ອງສະຫມັກ).

3. ໂດຍທົ່ວໄປ, ຕົວແທນສາມາດສ້າງຫົວຂໍ້ທີ່ມີການຄຸ້ມຄອງທີ່ມີຄ່າທົ່ວໂລກ, ແນວໃດກໍ່ຕາມ, ຂ້ອຍຢາກປະກາດທຸກຢ່າງຢ່າງຈະແຈ້ງ. ນອກຈາກນັ້ນ, ບາງພາລາມິເຕີ (ຕົວຢ່າງ, ຈໍານວນຂອງການແບ່ງສ່ວນຫຼືນະໂຍບາຍການຮັກສາໄວ້) ຂອງຫົວຂໍ້ໃນການໂຄສະນາຕົວແທນບໍ່ສາມາດຖືກຕັ້ງຄ່າໄດ້.

   ນີ້ແມ່ນສິ່ງທີ່ມັນອາດຈະເບິ່ງຄືວ່າໂດຍບໍ່ມີການກໍານົດຫົວຂໍ້ດ້ວຍຕົນເອງ:

app = get_app()

@app.agent()
async def collect_securities(stream: StreamT[None]) -> AsyncIterable[bool]:
	pass

ແລ້ວ, ຕອນນີ້ໃຫ້ອະທິບາຍສິ່ງທີ່ຕົວແທນຂອງພວກເຮົາຈະເຮັດ :)

app = get_app()

collect_securities_topic = app.topic("collect_securities", internal=True)

@app.agent(collect_securities_topic)
async def collect_securities(stream: StreamT[None]) -> AsyncIterable[bool]:
    async with aiohttp.ClientSession() as session:
        async for _ in stream:
            logger.info("Start collect securities")

            client = AlphaVantageClient(session, API_KEY)

            securities = await client.get_securities()

            for security in securities:
                await SecurityCRUD.update_one(
                    {"symbol": security["symbol"], "exchange": security["exchange"]}, security, upsert=True
                )

            yield True

ດັ່ງນັ້ນ, ໃນຕອນເລີ່ມຕົ້ນຂອງຕົວແທນ, ພວກເຮົາເປີດກອງປະຊຸມ aiohttp ສໍາລັບການຮ້ອງຂໍຜ່ານລູກຄ້າຂອງພວກເຮົາ. ດັ່ງນັ້ນ, ເມື່ອເລີ່ມຕົ້ນພະນັກງານ, ເມື່ອຕົວແທນຂອງພວກເຮົາຖືກເປີດຕົວ, ກອງປະຊຸມຈະເປີດທັນທີ - ຫນຶ່ງ, ສໍາລັບເວລາທີ່ພະນັກງານເຮັດວຽກທັງຫມົດ (ຫຼືຫຼາຍໆຄົນ, ຖ້າທ່ານປ່ຽນພາລາມິເຕີ. ພ້ອມກັນ ຈາກຕົວແທນທີ່ມີຫນ່ວຍງານເລີ່ມຕົ້ນ).

ຕໍ່ໄປ, ພວກເຮົາປະຕິບັດຕາມກະແສ (ພວກເຮົາວາງຂໍ້ຄວາມໃນ _, ເນື່ອງຈາກວ່າພວກເຮົາ, ໃນຕົວແທນນີ້, ບໍ່ສົນໃຈເນື້ອຫາ) ຂອງຂໍ້ຄວາມຈາກຫົວຂໍ້ຂອງພວກເຮົາ, ຖ້າພວກເຂົາມີຢູ່ໃນການຊົດເຊີຍໃນປະຈຸບັນ, ຖ້າບໍ່ດັ່ງນັ້ນວົງຈອນຂອງພວກເຮົາຈະລໍຖ້າການມາຮອດຂອງພວກເຂົາ. ດີ, ພາຍໃນ loop ຂອງພວກເຮົາ, ພວກເຮົາເຂົ້າສູ່ລະບົບການຮັບຂໍ້ຄວາມ, ໄດ້ຮັບບັນຊີລາຍຊື່ຂອງການເຄື່ອນໄຫວ (get_securities ກັບຄືນມາພຽງແຕ່ການເຄື່ອນໄຫວໂດຍຄ່າເລີ່ມຕົ້ນ, ເບິ່ງລະຫັດລູກຄ້າ) ຫຼັກຊັບແລະບັນທຶກມັນໄວ້ໃນຖານຂໍ້ມູນ, ກວດເບິ່ງວ່າມີຄວາມປອດໄພທີ່ມີ ticker ດຽວກັນແລະ. ການແລກປ່ຽນໃນຖານຂໍ້ມູນ, ຖ້າຫາກວ່າມີ, ຫຼັງຈາກນັ້ນມັນ (ເຈ້ຍ) ຈະໄດ້ຮັບການປັບປຸງພຽງແຕ່.

ເປີດຕົວການສ້າງຂອງພວກເຮົາ!

> docker-compose up -d
... Запуск контейнеров ...
> faust -A horton.agents worker --without-web -l info

ຄຸນນະສົມບັດ PS ອົງປະກອບເວັບ ຂ້າພະເຈົ້າຈະບໍ່ພິຈາລະນາ faust ໃນບົດຄວາມ, ດັ່ງນັ້ນພວກເຮົາກໍານົດທຸງທີ່ເຫມາະສົມ.

ໃນຄໍາສັ່ງເປີດຕົວຂອງພວກເຮົາ, ພວກເຮົາໄດ້ບອກ faust ບ່ອນທີ່ຈະຊອກຫາຈຸດປະສົງຂອງແອັບພລິເຄຊັນແລະສິ່ງທີ່ຕ້ອງເຮັດກັບມັນ (ເປີດຕົວພະນັກງານ) ກັບລະດັບຜົນຜະລິດບັນທຶກຂໍ້ມູນ. ພວກເຮົາໄດ້ຮັບຜົນຜະລິດດັ່ງຕໍ່ໄປນີ້:

Spoiler

┌ƒaµS† v1.10.4┬───────────────────────────────────────────────────┐
│ id          │ horton                                            │
│ transport   │ [URL('kafka://localhost:9092')]                   │
│ store       │ memory:                                           │
│ log         │ -stderr- (info)                                   │
│ pid         │ 1271262                                           │
│ hostname    │ host-name                                         │
│ platform    │ CPython 3.8.2 (Linux x86_64)                      │
│ drivers     │                                                   │
│   transport │ aiokafka=1.1.6                                    │
│   web       │ aiohttp=3.6.2                                     │
│ datadir     │ /path/to/project/horton-data                      │
│ appdir      │ /path/to/project/horton-data/v1                   │
└─────────────┴───────────────────────────────────────────────────┘
... логи, логи, логи ...

┌Topic Partition Set─────────┬────────────┐
│ topic                      │ partitions │
├────────────────────────────┼────────────┤
│ collect_securities         │ {0-7}      │
│ horton-__assignor-__leader │ {0}        │
└────────────────────────────┴────────────┘ 

ມັນມີຊີວິດຢູ່!!!

ໃຫ້​ເຮົາ​ເບິ່ງ​ທີ່​ກໍາ​ນົດ​ໄວ້​ແບ່ງ​ປັນ​. ດັ່ງທີ່ພວກເຮົາສາມາດເຫັນໄດ້, ຫົວຂໍ້ຖືກສ້າງຂື້ນດ້ວຍຊື່ທີ່ພວກເຮົາກໍານົດໄວ້ໃນລະຫັດ, ຈໍານວນພາທິຊັນເລີ່ມຕົ້ນ (8, ເອົາມາຈາກ. topic_partitions - ຕົວກໍານົດການວັດຖຸຂອງຄໍາຮ້ອງສະຫມັກ), ເນື່ອງຈາກວ່າພວກເຮົາບໍ່ໄດ້ລະບຸຄ່າສ່ວນບຸກຄົນສໍາລັບຫົວຂໍ້ຂອງພວກເຮົາ (ຜ່ານພາທິຊັນ). ຕົວແທນທີ່ເປີດຕົວໃນຜູ້ອອກແຮງງານແມ່ນໄດ້ຖືກມອບຫມາຍໃຫ້ທັງຫມົດ 8 ການແບ່ງປັນ, ເນື່ອງຈາກວ່າມັນເປັນພຽງແຕ່ຫນຶ່ງ, ແຕ່ນີ້ຈະຖືກປຶກສາຫາລືໃນລາຍລະອຽດເພີ່ມເຕີມໃນສ່ວນກ່ຽວກັບການຈັດກຸ່ມ.

ແລ້ວ, ຕອນນີ້ພວກເຮົາສາມາດໄປຫາປ່ອງຢ້ຽມປາຍທາງອື່ນແລະສົ່ງຂໍ້ຄວາມຫວ່າງໄປຫາຫົວຂໍ້ຂອງພວກເຮົາ:

> faust -A horton.agents send @collect_securities
{"topic": "collect_securities", "partition": 6, "topic_partition": ["collect_securities", 6], "offset": 0, "timestamp": ..., "timestamp_type": 0}

PS ໃຊ້ @ ພວກເຮົາສະແດງໃຫ້ເຫັນວ່າພວກເຮົາກໍາລັງສົ່ງຂໍ້ຄວາມໄປຫາຫົວຂໍ້ທີ່ມີຊື່ວ່າ "collect_securities".

ໃນ​ກໍ​ລະ​ນີ​ນີ້​, ຂໍ້​ຄວາມ​ໄດ້​ໄປ​ຫາ​ພາ​ທິ​ຊັນ 6 - ທ່ານ​ສາ​ມາດ​ກວດ​ສອບ​ການ​ນີ້​ໂດຍ​ການ​ໄປ kafdrop on​ localhost:9000

ໄປຫາປ່ອງຢ້ຽມຢູ່ປາຍຍອດກັບພະນັກງານຂອງພວກເຮົາ, ພວກເຮົາຈະເຫັນຂໍ້ຄວາມທີ່ມີຄວາມສຸກທີ່ຖືກສົ່ງໄປໂດຍໃຊ້ loguru:

2020-09-23 00:26:37.304 | INFO     | horton.agents:collect_securities:40 - Start collect securities

ພວກເຮົາຍັງສາມາດເບິ່ງເຂົ້າໄປໃນ mongo (ໃຊ້ Robo3T ຫຼື Studio3T) ແລະເບິ່ງວ່າຫຼັກຊັບແມ່ນຢູ່ໃນຖານຂໍ້ມູນ:

ຂ້າພະເຈົ້າບໍ່ແມ່ນມະຫາເສດຖີ, ແລະດັ່ງນັ້ນພວກເຮົາມີຄວາມພໍໃຈກັບຕົວເລືອກການເບິ່ງທໍາອິດ.

ຄວາມສຸກແລະຄວາມສຸກ - ຕົວແທນທໍາອິດແມ່ນກຽມພ້ອມ :)

ຕົວແທນພ້ອມແລ້ວ ຕົວແທນໃໝ່ ອາຍຸຍືນ!

ແມ່ນແລ້ວ, ຜູ້ຊາຍ, ພວກເຮົາໄດ້ກວມເອົາພຽງແຕ່ 1/3 ຂອງເສັ້ນທາງທີ່ກະກຽມໂດຍບົດຄວາມນີ້, ແຕ່ຢ່າທໍ້ຖອຍໃຈ, ເພາະວ່າໃນປັດຈຸບັນມັນຈະງ່າຍຂຶ້ນ.

ດັ່ງນັ້ນໃນປັດຈຸບັນພວກເຮົາຕ້ອງການຕົວແທນທີ່ເກັບກໍາຂໍ້ມູນ meta ແລະເຮັດໃຫ້ມັນເຂົ້າໄປໃນເອກະສານການລວບລວມ:

collect_security_overview_topic = app.topic("collect_security_overview", internal=True)


@app.agent(collect_security_overview_topic)
async def collect_security_overview(
    stream: StreamT[?],
) -> AsyncIterable[bool]:
    async with aiohttp.ClientSession() as session:
        async for event in stream:
            ...

ເນື່ອງຈາກຕົວແທນນີ້ຈະປະມວນຜົນຂໍ້ມູນກ່ຽວກັບຄວາມປອດໄພສະເພາະ, ພວກເຮົາຈໍາເປັນຕ້ອງຊີ້ບອກ ticker (ສັນຍາລັກ) ຂອງຄວາມປອດໄພນີ້ໃນຂໍ້ຄວາມ. ສໍາລັບຈຸດປະສົງນີ້ໃນ faust ມີ ການບັນທຶກ — ຫ້ອງ​ຮຽນ​ທີ່​ປະ​ກາດ​ໂຄງ​ການ​ຂໍ້​ຄວາມ​ໃນ​ຫົວ​ຂໍ້​ຕົວ​ແທນ​.

ໃນກໍລະນີນີ້, ໃຫ້ໄປ records.py ແລະອະທິບາຍວ່າຂໍ້ຄວາມຂອງຫົວຂໍ້ນີ້ຄວນຈະເປັນແນວໃດ:

import faust


class CollectSecurityOverview(faust.Record):
    symbol: str
    exchange: str

ດັ່ງທີ່ທ່ານອາດຈະໄດ້ຄາດເດົາ, faust ໃຊ້ຄໍາບັນຍາຍປະເພດ python ເພື່ອອະທິບາຍຮູບແບບຂໍ້ຄວາມ, ເຊິ່ງແມ່ນເຫດຜົນທີ່ວ່າສະບັບຕໍາ່ສຸດທີ່ສະຫນັບສະຫນຸນໂດຍຫ້ອງສະຫມຸດແມ່ນ. 3.6.

ໃຫ້ກັບຄືນໄປຫາຕົວແທນ, ກໍານົດປະເພດແລະເພີ່ມມັນ:

collect_security_overview_topic = app.topic(
    "collect_security_overview", internal=True, value_type=CollectSecurityOverview
)


@app.agent(collect_security_overview_topic)
async def collect_security_overview(
    stream: StreamT[CollectSecurityOverview],
) -> AsyncIterable[bool]:
    async with aiohttp.ClientSession() as session:
        async for event in stream:
            logger.info(
                "Start collect security [{symbol}] overview", symbol=event.symbol
            )

            client = AlphaVantageClient(session, API_KEY)

            security_overview = await client.get_security_overview(event.symbol)

            await SecurityCRUD.update_one({"symbol": event.symbol, "exchange": event.exchange}, security_overview)

            yield True

ດັ່ງທີ່ເຈົ້າສາມາດເຫັນໄດ້, ພວກເຮົາຜ່ານຕົວກໍານົດການໃຫມ່ທີ່ມີໂຄງການກັບວິທີການເລີ່ມຕົ້ນຫົວຂໍ້ - value_type . ນອກຈາກນັ້ນ, ທຸກສິ່ງທຸກຢ່າງປະຕິບັດຕາມໂຄງການດຽວກັນ, ສະນັ້ນຂ້າພະເຈົ້າບໍ່ເຫັນຈຸດໃດທີ່ຈະຢູ່ກັບສິ່ງອື່ນ.

ດີ, ການສໍາພັດສຸດທ້າຍແມ່ນການເພີ່ມການໂທຫາຕົວແທນເກັບກໍາຂໍ້ມູນ meta ເພື່ອ collect_securitites:

....
for security in securities:
    await SecurityCRUD.update_one({
            "symbol": security["symbol"],
            "exchange": security["exchange"]
        },
        security,
        upsert = True,
    )

    await collect_security_overview.cast(
        CollectSecurityOverview(symbol = security["symbol"], exchange = security["exchange"])
    )
....

ພວກເຮົານໍາໃຊ້ໂຄງການທີ່ປະກາດກ່ອນຫນ້ານີ້ສໍາລັບຂໍ້ຄວາມ. ໃນກໍລະນີນີ້, ຂ້າພະເຈົ້າໄດ້ໃຊ້ວິທີການ .cast ນັບຕັ້ງແຕ່ພວກເຮົາບໍ່ຈໍາເປັນຕ້ອງລໍຖ້າຜົນໄດ້ຮັບຈາກຕົວແທນ, ແຕ່ມັນສົມຄວນທີ່ຈະກ່າວເຖິງວ່າ. ທາງ ສົ່ງຂໍ້ຄວາມໄປຫາຫົວຂໍ້:

1. cast - ບໍ່ຕັນເນື່ອງຈາກວ່າມັນບໍ່ໄດ້ຄາດຫວັງຜົນໄດ້ຮັບ. ທ່ານບໍ່ສາມາດສົ່ງຜົນໄດ້ຮັບໄປຫາຫົວຂໍ້ອື່ນເປັນຂໍ້ຄວາມ.

2. send - ບໍ່ຕັນເພາະວ່າມັນບໍ່ໄດ້ຄາດຫວັງຜົນໄດ້ຮັບ. ທ່ານສາມາດລະບຸຕົວແທນໃນຫົວຂໍ້ທີ່ຜົນໄດ້ຮັບຈະໄປ.

3. ຖາມ - ລໍຖ້າຜົນໄດ້ຮັບ. ທ່ານສາມາດລະບຸຕົວແທນໃນຫົວຂໍ້ທີ່ຜົນໄດ້ຮັບຈະໄປ.

ດັ່ງນັ້ນ, ທັງຫມົດທີ່ມີຕົວແທນສໍາລັບມື້ນີ້!

ທີມງານຝັນ

ສິ່ງສຸດທ້າຍທີ່ຂ້ອຍສັນຍາວ່າຈະຂຽນໃນສ່ວນນີ້ແມ່ນຄໍາສັ່ງ. ດັ່ງທີ່ໄດ້ກ່າວກ່ອນຫນ້ານີ້, ຄໍາສັ່ງໃນ faust ແມ່ນ wrapper ປະມານຄລິກ. ໃນຄວາມເປັນຈິງ, faust ພຽງແຕ່ເອົາຄໍາສັ່ງທີ່ກໍາຫນົດເອງຂອງພວກເຮົາໃສ່ໃນການໂຕ້ຕອບຂອງມັນໃນເວລາທີ່ກໍານົດລະຫັດ -A

ຫຼັງຈາກຕົວແທນປະກາດໃນ agents.py ເພີ່ມຟັງຊັນທີ່ມີເຄື່ອງຕົບແຕ່ງ app.commandໂທຫາວິທີການ ແມ່ພິມສໍາລັບ у collect_securitites:

@app.command()
async def start_collect_securities():
    """Collect securities and overview."""

    await collect_securities.cast()

ດັ່ງນັ້ນ, ຖ້າພວກເຮົາໂທຫາບັນຊີລາຍຊື່ຂອງຄໍາສັ່ງ, ຄໍາສັ່ງໃຫມ່ຂອງພວກເຮົາຈະຢູ່ໃນມັນ:

> faust -A horton.agents --help

....
Commands:
  agents                    List agents.
  clean-versions            Delete old version directories.
  completion                Output shell completion to be evaluated by the...
  livecheck                 Manage LiveCheck instances.
  model                     Show model detail.
  models                    List all available models as a tabulated list.
  reset                     Delete local table state.
  send                      Send message to agent/topic.
  start-collect-securities  Collect securities and overview.
  tables                    List available tables.
  worker                    Start worker instance for given app.

ພວກເຮົາສາມາດໃຊ້ມັນຄືກັບຜູ້ອື່ນ, ສະນັ້ນໃຫ້ພວກເຮົາເລີ່ມພະນັກງານ faust ໃໝ່ ແລະເລີ່ມຕົ້ນການເກັບຫຼັກຊັບທີ່ເຕັມຮູບແບບ:

> faust -A horton.agents start-collect-securities

ຈະມີຫຍັງເກີດຂື້ນຕໍ່ໄປ?

ໃນສ່ວນຕໍ່ໄປ, ການນໍາໃຊ້ຕົວແທນທີ່ຍັງເຫຼືອເປັນຕົວຢ່າງ, ພວກເຮົາຈະພິຈາລະນາກົນໄກການຫລົ້ມຈົມໃນການຄົ້ນຫາທີ່ສຸດໃນລາຄາປິດຂອງການຊື້ຂາຍສໍາລັບປີແລະການເປີດຕົວ cron ຂອງຕົວແທນ.

ນັ້ນແມ່ນທັງໝົດສຳລັບມື້ນີ້! ຂອບໃຈສໍາລັບການອ່ານ :)

ລະຫັດສໍາລັບສ່ວນນີ້

PS ພາຍ​ໃຕ້​ພາກ​ສ່ວນ​ສຸດ​ທ້າຍ​ຂ້າ​ພະ​ເຈົ້າ​ໄດ້​ຖືກ​ຖາມ​ກ່ຽວ​ກັບ faust ແລະ confluent kafka (confluent ມີຄຸນສົມບັດແນວໃດ?). ມັນເບິ່ງຄືວ່າ confluent ມີປະໂຫຍດຫຼາຍໃນຫຼາຍວິທີ, ແຕ່ຄວາມຈິງແລ້ວແມ່ນວ່າ faust ບໍ່ໄດ້ຮັບການສະຫນັບສະຫນູນລູກຄ້າຢ່າງເຕັມທີ່ສໍາລັບ confluent - ນີ້ແມ່ນມາຈາກ. ຄໍາອະທິບາຍກ່ຽວກັບຂໍ້ຈໍາກັດຂອງລູກຄ້າໃນເອກະສານ.

ແຫຼ່ງຂໍ້ມູນ: www.habr.com