Python Archives - Scienza dei Dati

Analisi delle Crypto Valute: Raccogliere i dati con Kafka, ClickHouse e Cryptofeed

scienzadeidati articoli - Raccogliere i dati con Kafka ClickHouse Cryptofeed

            version: '3.9'

services: 

  clickhouse-server:
    image: yandex/clickhouse-server
    ports: 
     - "8123:8123"
     - "8124:9000"
    volumes:
      - clickhouse-volume:/var/lib/clickhouse
  
  cryptofeed:
    build:
      context: .
      dockerfile: Dockerfile
    volumes:
      - ./:/app
    depends_on:
    - zookeeper
    - kafka

  zookeeper:
    image: wurstmeister/zookeeper:3.4.6
    expose:
    - "2181"

  kafka:
    image: wurstmeister/kafka:2.11-2.0.0
    depends_on:
    - zookeeper
    ports:
    - "9092:9092"
    expose:
    - "9093"
    environment:
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
     
volumes:
  clickhouse-volume:

            
from cryptofeed import FeedHandler
from cryptofeed.backends.kafka import BookKafka, TradeKafka
from cryptofeed.defines import L2_BOOK, TRADES, L3_BOOK
from cryptofeed.exchanges import Coinbase, Binance, Bitfinex


def main():
    f = FeedHandler()
    HOSTNAME = 'kafka'
    cbs = {TRADES: TradeKafka(bootstrap=HOSTNAME), L2_BOOK: BookKafka(bootstrap=HOSTNAME), L3_BOOK: BookKafka(bootstrap=HOSTNAME)}

    # Add trade and lv 2 bitcoin data to Feed
    f.add_feed(Bitfinex(max_depth=25, channels=[TRADES, L2_BOOK], symbols=['BTC-USD'], callbacks=cbs))
    f.add_feed(Binance(max_depth=25, channels=[TRADES, L2_BOOK], symbols=['BTC-BUSD'], callbacks=cbs))
    
    # Example of how to extract level 3 order book data
    # f.add_feed(Coinbase(max_depth=25, channels=[TRADES, L2_BOOK, L3_BOOK], symbols=['BTC-USD'], callbacks=cbs))

    f.run()


if __name__ == '__main__':
    main()

            FROM python:3.7-slim

WORKDIR /app
COPY . .

# upgrade pip, install requirements and update permissions for main.py
RUN pip install --upgrade pip \
    && pip install -r ./requirements.txt \
    && chmod +x /app/main.py

# Create CryptoFeed Kafka Producers
CMD ["python", "/app/main.py"]

            CREATE TABLE CryptoCrypt.Trades (
        feed String,
        symbol String,
        trade_date DateTime64(3),
        receipt_timestamp DateTime64(3),
        side String,
        amount Float64,
        price Float64,
        order_type Nullable(String), 
        id UInt64
        ) Engine = MergeTree
PARTITION BY toYYYYMM(trade_date)
ORDER BY (feed, trade_date);


CREATE TABLE CryptoCrypt.L2_orderbook (
        trade_date DateTime64(3),
        receipt_timestamp DateTime64(3),
        delta Int8,
        bid Map(String, String),
        ask Map(String, String)
        ) Engine = MergeTree
PARTITION BY toYYYYMM(trade_date)
ORDER BY (trade_date);

ALTER TABLE CryptoCrypt.L2_orderbook
  ADD COLUMN _topic String

            CREATE TABLE CryptoCrypt.trades_queue (
        feed String,
        symbol String,
        trade_date DateTime64(3),
        receipt_timestamp Float64,
        side String,
        amount Float64,
        price Float64,
        order_type Nullable(String), 
        id UInt64
        ) ENGINE = Kafka
SETTINGS kafka_broker_list = 'kafka:9092',
       kafka_topic_list = 'trades-BINANCE-BTC-BUSD, trades-BITFINEX-BTC-USD',
       kafka_group_name = 'trades_group1',
       kafka_format = 'JSONEachRow',
       kafka_row_delimiter = '\n';

CREATE TABLE CryptoCrypt.books_queue (
        trade_date Float64,
        receipt_timestamp Float64,
        delta Int8,
        bid Map(String, LowCardinality(Float64)),
        ask Map(String, LowCardinality(Float64))
        ) ENGINE = Kafka
SETTINGS kafka_broker_list = 'kafka:9092',
       kafka_topic_list = 'book-BINANCE-BTC-BUSD, book-BITFINEX-BTC-USD',
       kafka_group_name = 'trades_group1',
       kafka_format = 'JSONEachRow',
       kafka_row_delimiter = '\n';

            CREATE MATERIALIZED VIEW CryptoCrypt.trades_queue_mv TO CryptoCrypt.Trades AS
SELECT *
FROM CryptoCrypt.trades_queue;

CREATE MATERIALIZED VIEW CryptoCrypt.books_queue_mv TO CryptoCrypt.L2_orderbook AS
SELECT trade_date, receipt_timestamp, delta, cast(bid, 'Map(String, String)') as bid , cast(ask, 'Map(String, String)') as ask, _topic 
FROM CryptoCrypt.books_queue;

            # Show all tables, should have 2 kafka, 2 MergeTree and 2 views
SHOW TABLES FROM CryptoCrypt

# Check count of MergeTree tables
SELECT COUNT() FROM CryptoCrypt.L2_orderbook
SELECT COUNT() FROM CryptoCrypt.Trades

# Check if data is correct format 
SELECT * FROM CryptoCrypt.L2_orderbook LIMIT 5
SELECT * FROM CryptoCrypt.Trades LIMIT 5

# If no data in MergeTree tables check Kafka tables
SELECT * FROM CryptoCrypt.books_queue LIMIT 5
SELECT * FROM CryptoCrypt.trades_queue LIMIT 5

            # From your terminal run: MAKE SURE TO SET KAFKA CONTAINER NAME
> docker exec -i -t -u root $(docker ps | grep NAME_OF_KAFKA_CONTAINER | cut -d' ' -f1) /bin/bash

# List all topics
bash> $KAFKA_HOME/bin/kafka-topics.sh --list --bootstrap-server kafka:9092

# Check if a topic is collecting messages 
bash> $KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic book-BITFINEX-BTC-USD --from-beginning

cryptocurrencies Python

Creare un’API di dati azionari utilizzando il Web Scraping e FastAPI

scienzadeidati articoli - Creare API di dati finanziari utilizzando Web Scraping e FastAPI

            {
    "elements": [
        {
            "from": "",
            "to": ""
        },
        {
            "from": "",
            "to": ""
        },
        {
            "from": "",
            "to": ""
        }
    ]
}

            {
    "elements": [
        {
            "from": "[class='Trsdu\\(0\\.3s\\) Fw\\(b\\) Fz\\(36px\\) Mb\\(-4px\\) D\\(ib\\)']",
            "to": "price"
        },
        {
            "from": "[data-test='PREV_CLOSE-value'] [class]",
            "to": "prev_close"
        },
        {
            "from": "[data-test='OPEN-value'] [class]",
            "to": "open"
        },
        {
            "from": "[data-test='BID-value'] [class]",
            "to": "bid"
        },
        {
            "from": "[data-test='ASK-value'] [class]",
            "to": "ask"
        },
        {
            "from": "[data-test='DAYS_RANGE-value']",
            "to": "days_range"
        },
        {
            "from": "[data-test='FIFTY_TWO_WK_RANGE-value']",
            "to": "52_week_range"
        },
        {
            "from": "[data-test='TD_VOLUME-value'] [class]",
            "to": "volume"
        },
        {
            "from": "[data-test='AVERAGE_VOLUME_3MONTH-value'] [class]",
            "to": "avg_volume"
        },
        {
            "from": "[data-test='MARKET_CAP-value'] [class]",
            "to": "market_cap"
        },
        {
            "from": "[data-test='BETA_5Y-value'] [class]",
            "to": "beta"
        },
        {
            "from": "[data-test='PE_RATIO-value'] [class]",
            "to": "pe"
        },
        {
            "from": "[data-test='EPS_RATIO-value'] [class]",
            "to": "eps"
        },
        {
            "from": "[data-test='EARNINGS_DATE-value']",
            "to": "earnings_date"
        },
        {
            "from": "[data-test='DIVIDEND_AND_YIELD-value']",
            "to": "dividend_and_yield"
        },
        {
            "from": "[data-test='EX_DIVIDEND_DATE-value'] span",
            "to": "ex_dividend_yield"
        },
        {
            "from": "[data-test='ONE_YEAR_TARGET_PRICE-value'] [class]",
            "to": "target_est"
        }
    ]
}

            import requests
from bs4 import BeautifulSoup

class Scrape():

    def __init__(self, symbol, elements):
        url = "https://finance.yahoo.com/quote/" + symbol

        r = requests.get(url)
        if(r.url != url): # redirect occurred; likely symbol doesn't exist or cannot be found.
            raise requests.TooManyRedirects()

        r.raise_for_status()
        
        self.soup = BeautifulSoup(r.text, "html.parser")

        self.__summary = {}

        for el in elements["elements"]:
            tag = self.soup.select_one(el["from"])

            if tag != None:
                self.__summary[el["to"]] = tag.get_text()

    def summary(self):
        return self.__summary

            from Scrape import Scrape
import json

elements_to_scrape = {}

f = open("scrape.json")
data = f.read()
f.close()
elements_to_scrape = json.loads(data)

s = Scrape("AAPL", elements_to_scrape)
print(s.summary())

            @app.get("/v1/{symbol}/summary/")
def summary(symbol):
    summary_data = {}
    try:
        s = Scrape(symbol, elements_to_scrape)
        summary_data = s.summary()
        
    except TooManyRedirects:
        raise HTTPException(status_code=404, 
              detail="{symbol} doesn't exist or cannot be found.")
    except HTTPError:
        raise HTTPException(status_code=500, 
              detail="An error has occurred while processing the request.")

    return summary_data

            @app.on_event("startup")
def startup():
    f = open("scrape.json")
    data = f.read()
    f.close()
    global elements_to_scrape
    elements_to_scrape = json.loads(data)

            {
  "price": "135.37",
  "prev_close": "135.13",
  "open": "134.35",
  "bid": "135.40 x 1000",
  "ask": "135.47 x 1300",
  "days_range": "133.69 - 135.51",
  "52_week_range": "53.15 - 145.09",
  "volume": "60,145,130",
  "avg_volume": "102,827,562",
  "market_cap": "2.273T",
  "beta": "1.27",
  "pe": "36.72",
  "eps": "3.69",
  "earnings_date": "Apr 28, 2021 - May 03, 2021",
  "dividend_and_yield": "0.82 (0.61%)",
  "ex_dividend_yield": "Feb 05, 2021",
  "target_est": "151.75"
}

            @app.get("/v1/{symbol}/summary/")
def summary(symbol, q: Optional[List[str]] = Query(None)):
    summary_data = {}
    try:
        s = Scrape(symbol, elements_to_scrape)
        summary_data = s.summary()
        if(q != None):
            # if all query parameters are keys in summary_data
            if all (k in summary_data for k in q): 
                # summary_data keeps requested key-value pairs
                summary_data = {key: summary_data[key] for key in q}
            # else, return whole summary_data

    except TooManyRedirects:
        raise HTTPException(status_code=404, detail="{symbol} doesn't exist or cannot be found.")
    except HTTPError:
        raise HTTPException(status_code=500, detail="An error has occurred while processing the request.")

    return summary_data

Web scraping automatizzato con Python e Celery

scienzadeidati articoli - Web scraping automatizzato Python Celery

				
					$ sudo apt-get update -y

$ sudo apt-get install curl gnupg -y

$ curl -fsSl https://github.com/rabbitmq/signing-keys/releases/download/2.0/rabbitmq-release-signing-key.asc | sudo apt-key add -

$ sudo apt-get install apt-transport https

$ sudo tee /etc/apt/sources.list.d/bintray.rabbitmq.list <<EOF

$ deb https://dl.bintray.com/rabbitmq-erlang/debian bionic erlang

$ deb https://dl.bintray.com/rabbitmq/debian bionic main

$ EOF

$ sudo apt-get update -y

$ sudo apt-get install rabbitmq-server -y --fix-missing

				
					# tasks.py

from celery import Celery

app = Celery('tasks') # defining the app name to be used in our flag

@app.task # registering the task to the app
def add(x, y):
    return x + y

				
					# tasks.py

from celery import Celery
import requests # pulling data
from bs4 import BeautifulSoup # xml parsing
import json # exporting to files

app = Celery('tasks')

# save function
def save_function(article_list):
    with open('articles.txt', 'w' as outfile:
        json.dump(article_list, outfile)
        
# scraping function
def hackernews_rss():
    article_list = []
    try:
        # execute my request, parse the data using the XML 
        # parser in BS4
        r = requests.get('https://news.ycombinator.com/rss')
        soup = BeautifulSoup(r.content, features='xml')
        # select only the "items" I want from the data      
        articles = soup.findAll('item')
        # for each "item" I want, parse it into a list
        for a in articles:
            title = a.find('title').text
            link = a.find('link').text
            published = a.find('pubDate').text
            # create an "article" object with the data
            # from each "item"
            article = {
                'title': title,
                'link': link,
                'published': published
                }
            # append my "article_list" with each "article" object
            article_list.append(article)
        # after the loop, dump my saved objects into a .txt file
        return save_function(article_list)
    except Exception as e:
        print('The scraping job failed. See exception: ')
        print(e)

				
					# tasks.py

from datetime import datetime # for time stamps

... 

def save_function(articles_list):
    # timestamp and filename
    timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
    filename = 'articles-{}.json'.format(timestamp)
    # creating our articles file with timestamp
    with open(filename, 'w').format(timestamp) as outfile:
        json.dump(article_list, outfile)

				
					# tasks.py

...

def hackernews_rss():
    ...
        for a in articles:
            ...
            article = {
                ...
                'created_at': str(datetime.now()),
                'source': 'HackerNews RSS'
                }
        ...

				
					# tasks.py

... 

from celery.schedules import crontab # scheduler

# scheduled task execution
app.conf.beat_schedule = {
    # executes every 1 minute
    'scraping-task-one-min': {
        'task': 'tasks.hackernews_rss',
        'schedule': crontab()
    }
}
...

Celery Python

Creazione di uno scraper di feed RSS con Python

scienzadeidati articoli - Costruire uno scraper di feed RSS con Python

            # scraping.py

# library imports
...

# scraping function
def hackernews_rss('https://news.ycombinator.com/rss'):
    try:
        r = requests.get()
        return print('The scraping job succeeded: ', r.status_code)
    except Exception as e:
        print('The scraping job failed. See exception: ')
        print(e)
        
        
print('Starting scraping')
hackernews_rss()
print('Finished scraping')

            # scraping.py

...

def hackernews_rss():
    try:
        r = requests.get('https://news.ycombinator.com/rss')
        soup = BeautifulSoup(r.content, features='xml')
        
        return print(soup)
...

            # scraping.py 

...

def hackernews_rss():
    article_list = []
    try:
        r = requests.get('https://news.ycombinator.com/rss')
        soup = BeautifulSoup(r.content, features='xml')
        articles = soup.findAll('item')        
        for a in articles:
            title = a.find('title').text
            link = a.find('link').text
            published = a.find('pubDate').text
            article = {
                'title': title,
                'link': link,
                'published': published
                }
            article_list.append(article)
        return print(article_list)
...

            # scraping.py

...

def save_function(article_list):
    with open('articles.txt', 'w') as f:
        for a in article_list:
            f.write(a+'\n')
        f.close()
        
...

Python Web Scraping

Creazione di Microservizi con Python e FastAPI

            #~/movie_service/app/main.py

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

fake_movie_db = [
    {
        'name': 'Star Wars: Episode IX - The Rise of Skywalker',
        'plot': 'The surviving members of the resistance face the First Order once again.',
        'genres': ['Action', 'Adventure', 'Fantasy'],
        'casts': ['Daisy Ridley', 'Adam Driver']
    }
]

class Movie(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts: List[str]


@app.get('/', response_model=List[Movie])
async def index():
    return fake_movie_db

            @app.post('/', status_code=201)
async def add_movie(payload: Movie):
    movie = payload.dict()
    fake_movie_db.append(movie)
    return {'id': len(fake_movie_db) - 1}

            @app.put('/{id}')
async def update_movie(id: int, payload: Movie):
    movie = payload.dict()
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        fake_movie_db[id] = movie
        return None
    raise HTTPException(status_code=404, 
                        detail="Movie with given id not found")

            @app.delete('/{id}')
async def delete_movie(id: int):
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        del fake_movie_db[id]
        return None
    raise HTTPException(status_code=404, 
                        detail="Movie with given id not found")

            #~/movie-service/app/api/movies.py

from typing import List
from fastapi import Header, APIRouter

from app.api.models import Movie

fake_movie_db = [
    {
        'name': 'Star Wars: Episode IX - The Rise of Skywalker',
        'plot': 'The surviving members of the resistance face the First Order once again.',
        'genres': ['Action', 'Adventure', 'Fantasy'],
        'casts': ['Daisy Ridley', 'Adam Driver']
    }
]

movies = APIRouter()

@movies.get('/', response_model=List[Movie])
async def index():
    return fake_movie_db

@movies.post('/', status_code=201)
async def add_movie(payload: Movie):
    movie = payload.dict()
    fake_movie_db.append(movie)
    return {'id': len(fake_movie_db) - 1}

@movies.put('/{id}')
async def update_movie(id: int, payload: Movie):
    movie = payload.dict()
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        fake_movie_db[id] = movie
        return None
    raise HTTPException(status_code=404, detail="Movie with given id not found")

@movies.delete('/{id}')
async def delete_movie(id: int):
    movies_length = len(fake_movie_db)
    if 0 <= id <= movies_length:
        del fake_movie_db[id]
        return None
    raise HTTPException(status_code=404, detail="Movie with given id not found")

            #~/movie-service/api/models.py

from typing import List
from pydantic import BaseModel

class Movie(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts: List[str]

            #~/movie-service/app/api/db.py

from sqlalchemy import (Column, Integer, MetaData, String, Table,
                        create_engine, ARRAY)

from databases import Database

DATABASE_URL = 'postgresql://movie_user:movie_password@localhost/movie_db'

engine = create_engine(DATABASE_URL)
metadata = MetaData()

movies = Table(
    'movies',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('plot', String(250)),
    Column('genres', ARRAY(String)),
    Column('casts', ARRAY(String))
)

database = Database(DATABASE_URL)

            #~/movie-service/app/main.py

from fastapi import FastAPI
from app.api.movies import movies
from app.api.db import metadata, database, engine

metadata.create_all(engine)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()
    
app.include_router(movies)

            #~/movie-service/app/api/movies.py

from typing import List
from fastapi import Header, APIRouter

from app.api.models import MovieIn, MovieOut
from app.api import db_manager

movies = APIRouter()

@movies.get('/', response_model=List[MovieOut])
async def index():
    return await db_manager.get_all_movies()

@movies.post('/', status_code=201)
async def add_movie(payload: MovieIn):
    movie_id = await db_manager.add_movie(payload)
    response = {
        'id': movie_id,
        **payload.dict()
    }

    return response

@movies.put('/{id}')
async def update_movie(id: int, payload: MovieIn):
    movie = payload.dict()
    fake_movie_db[id] = movie
    return None

@movies.put('/{id}')
async def update_movie(id: int, payload: MovieIn):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")

    update_data = payload.dict(exclude_unset=True)
    movie_in_db = MovieIn(**movie)

    updated_movie = movie_in_db.copy(update=update_data)

    return await db_manager.update_movie(id, updated_movie)

@movies.delete('/{id}')
async def delete_movie(id: int):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")
    return await db_manager.delete_movie(id)

            #~/movie-service/app/api/db_manager.py

from app.api.models import MovieIn, MovieOut, MovieUpdate
from app.api.db import movies, database

async def add_movie(payload: MovieIn):
    query = movies.insert().values(**payload.dict())
    return await database.execute(query=query)

async def get_all_movies():
    query = movies.select()
    return await database.fetch_all(query=query)

async def get_movie(id):
    query = movies.select(movies.c.id==id)
    return await database.fetch_one(query=query)

async def delete_movie(id: int):
    query = movies.delete().where(movies.c.id==id)
    return await database.execute(query=query)

async def update_movie(id: int, payload: MovieIn):
    query = (
        movies
        .update()
        .where(movies.c.id == id)
        .values(**payload.dict())
    )
    return await database.execute(query=query)

            #~/movie-service/app/api/models.py

from pydantic import BaseModel
from typing import List, Optional


class MovieIn(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts: List[str]


class MovieOut(MovieIn):
    id: int


class MovieUpdate(MovieIn):
    name: Optional[str] = None
    plot: Optional[str] = None
    genres: Optional[List[str]] = None
    casts: Optional[List[str]] = None

            FROM python:3.8-slim

WORKDIR /app

COPY ./requirements.txt /app/requirements.txt

RUN apt-get update \
    && apt-get install gcc -y \
    && apt-get clean

RUN pip install -r /app/requirements.txt \
    && rm -rf /root/.cache/pip

COPY . /app/

            #~/python-microservices/movie-service/app/api/models.py

from pydantic import BaseModel
from typing import List, Optional

class MovieIn(BaseModel):
    name: str
    plot: str
    genres: List[str]
    casts_id: List[int]


class MovieOut(MovieIn):
    id: int


class MovieUpdate(MovieIn):
    name: Optional[str] = None
    plot: Optional[str] = None
    genres: Optional[List[str]] = None
    casts_id: Optional[List[int]] = None

            #~/python-microservices/movie-service/app/api/db.py

import os

from sqlalchemy import (Column, DateTime, Integer, MetaData, String, Table,
                        create_engine, ARRAY)

from databases import Database

DATABASE_URL = os.getenv('DATABASE_URL')

engine = create_engine(DATABASE_URL)
metadata = MetaData()

movies = Table(
    'movies',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('plot', String(250)),
    Column('genres', ARRAY(String)),
    Column('casts_id', ARRAY(Integer))
)

database = Database(DATABASE_URL)

            #~/python-microservices/movie-service/app/api/movies.py

from typing import List
from fastapi import APIRouter, HTTPException

from app.api.models import MovieOut, MovieIn, MovieUpdate
from app.api import db_manager
from app.api.service import is_cast_present

movies = APIRouter()

@movies.post('/', response_model=MovieOut, status_code=201)
async def create_movie(payload: MovieIn):
    for cast_id in payload.casts_id:
        if not is_cast_present(cast_id):
            raise HTTPException(status_code=404, detail=f"Cast with id:{cast_id} not found")

    movie_id = await db_manager.add_movie(payload)
    response = {
        'id': movie_id,
        **payload.dict()
    }

    return response

@movies.get('/', response_model=List[MovieOut])
async def get_movies():
    return await db_manager.get_all_movies()

@movies.get('/{id}/', response_model=MovieOut)
async def get_movie(id: int):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")
    return movie

@movies.put('/{id}/', response_model=MovieOut)
async def update_movie(id: int, payload: MovieUpdate):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")

    update_data = payload.dict(exclude_unset=True)

    if 'casts_id' in update_data:
        for cast_id in payload.casts_id:
            if not is_cast_present(cast_id):
                raise HTTPException(status_code=404, detail=f"Cast with given id:{cast_id} not found")

    movie_in_db = MovieIn(**movie)

    updated_movie = movie_in_db.copy(update=update_data)

    return await db_manager.update_movie(id, updated_movie)

@movies.delete('/{id}', response_model=None)
async def delete_movie(id: int):
    movie = await db_manager.get_movie(id)
    if not movie:
        raise HTTPException(status_code=404, detail="Movie not found")
    return await db_manager.delete_movie(id)

            #~/python-microservices/movie-service/app/api/service.py

import os
import httpx

CAST_SERVICE_HOST_URL = 'http://localhost:8002/api/v1/casts/'
url = os.environ.get('CAST_SERVICE_HOST_URL') or CAST_SERVICE_HOST_URL

def is_cast_present(cast_id: int):
    r = httpx.get(f'{url}{cast_id}')
    return True if r.status_code == 200 else False

            python-microservices/
.
├── cast_service/
│   ├── app/
│   │   ├── api/
│   │   │   ├── casts.py
│   │   │   ├── db_manager.py
│   │   │   ├── db.py
│   │   │   ├── models.py
│   │   ├── main.py
│   ├── Dockerfile
│   └── requirements.txt
├── movie_service/
...

            FROM python:3.8-slim

WORKDIR /app

COPY ./requirements.txt /app/requirements.txt

RUN apt-get update \
    && apt-get install gcc -y \
    && apt-get clean

RUN pip install -r /app/requirements.txt \
    && rm -rf /root/.cache/pip

COPY . /app/

            #~/python-microservices/cast-service/app/main.py

from fastapi import FastAPI
from app.api.casts import casts
from app.api.db import metadata, database, engine

metadata.create_all(engine)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

app.include_router(casts, prefix='/api/v1/casts', tags=['casts'])

            #~/python-microservices/cast-service/app/api/casts.py

from fastapi import APIRouter, HTTPException
from typing import List

from app.api.models import CastOut, CastIn, CastUpdate
from app.api import db_manager

casts = APIRouter()

@casts.post('/', response_model=CastOut, status_code=201)
async def create_cast(payload: CastIn):
    cast_id = await db_manager.add_cast(payload)

    response = {
        'id': cast_id,
        **payload.dict()
    }

    return response

@casts.get('/{id}/', response_model=CastOut)
async def get_cast(id: int):
    cast = await db_manager.get_cast(id)
    if not cast:
        raise HTTPException(status_code=404, detail="Cast not found")
    return cast

            #~/python-microservices/cast-service/app/api/db_manager.py

from app.api.models import CastIn, CastOut, CastUpdate
from app.api.db import casts, database


async def add_cast(payload: CastIn):
    query = casts.insert().values(**payload.dict())

    return await database.execute(query=query)

async def get_cast(id):
    query = casts.select(casts.c.id==id)
    return await database.fetch_one(query=query)

            #~/python-microservices/cast-service/app/api/db.py

import os

from sqlalchemy import (Column, Integer, MetaData, String, Table,
                        create_engine, ARRAY)

from databases import Database

DATABASE_URI = os.getenv('DATABASE_URI')

engine = create_engine(DATABASE_URI)
metadata = MetaData()

casts = Table(
    'casts',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('nationality', String(20)),
)

database = Database(DATABASE_URI)

            #~/python-microservices/cast-service/app/api/models.py

from pydantic import BaseModel
from typing import List, Optional

class CastIn(BaseModel):
    name: str
    nationality: Optional[str] = None


class CastOut(CastIn):
    id: int


class CastUpdate(CastIn):
    name: Optional[str] = None

            version: '3.7'

services:
  movie_service:
    build: ./movie-service
    command: uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
    volumes:
      - ./movie-service/:/app/
    ports:
      - 8001:8000
    environment:
      - DATABASE_URI=postgresql://movie_db_username:movie_db_password@movie_db/movie_db_dev
      - CAST_SERVICE_HOST_URL=http://cast_service:8000/api/v1/casts/

  movie_db:
    image: postgres:12.1-alpine
    volumes:
      - postgres_data_movie:/var/lib/postgresql/data/
    environment:
      - POSTGRES_USER=movie_db_username
      - POSTGRES_PASSWORD=movie_db_password
      - POSTGRES_DB=movie_db_dev

  cast_service:
    build: ./cast-service
    command: uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
    volumes:
      - ./cast-service/:/app/
    ports:
      - 8002:8000
    environment:
      - DATABASE_URI=postgresql://cast_db_username:cast_db_password@cast_db/cast_db_dev

  cast_db:
    image: postgres:12.1-alpine
    volumes:
      - postgres_data_cast:/var/lib/postgresql/data/
    environment:
      - POSTGRES_USER=cast_db_username
      - POSTGRES_PASSWORD=cast_db_password
      - POSTGRES_DB=cast_db_dev

volumes:
  postgres_data_movie:
  postgres_data_cast:

            server {
  listen 8080;

  location /api/v1/movies {
    proxy_pass http://movie_service:8000/api/v1/movies;
  }

  location /api/v1/casts {
    proxy_pass http://cast_service:8000/api/v1/casts;
  }

}

            ...
nginx:
    image: nginx:latest
    ports:
      - "8080:8080"
    volumes:
      - ./nginx_config.conf:/etc/nginx/conf.d/default.conf
    depends_on:
      - cast_service
      - movie_service
...

FastAPI Microservices Python

Pianificare un web scraping con Apache Airflow

scienzadeidati articoli - Web Scarping con Airflow

            import datetime as dt

from airflow import DAG
from airflow.operators.python_operator import PythonOperator


def parse_recipes():
    return 'parse_recipes'


def download_image():
    return 'download_image'


def store_data():
    return 'store_data'


default_args = {
    'owner': 'airflow',
    'start_date': dt.datetime(2020, 12, 1, 10, 00, 00),
    'concurrency': 1,
    'retries': 0
}

with DAG('parsing_recipes',
         catchup=False,
         default_args=default_args,
         schedule_interval='*/10 * * * *',
         # schedule_interval=None,
         ) as dag:
    opr_parse_recipes = PythonOperator(task_id='parse_recipes',
                                       python_callable=parse_recipes)

    opr_download_image = PythonOperator(task_id='download_image',
                                        python_callable=download_image)
    opr_store_data = PythonOperator(task_id='store_data',
                                    python_callable=store_data)

opr_parse_recipes >> opr_download_image >> opr_store_data

            
def download_image(**kwargs):
    ti = kwargs['ti']
    v1 = ti.xcom_pull(key=None, task_ids='parse_recipes')

    print('Printing Task 1 values in Download_image')
    print(v1)
    return 'download_image'

            {
  'dag': <DAG: parsing_recipes>,
  'ds': '2020-12-02',
  'next_ds': '2020-12-02',
  'prev_ds': '2020-12-02',
  'ds_nodash': '20201202',
  'ts': '2020-12-02T09:56:05.289457+00:00',
  'ts_nodash': '20201202T095605.289457+0000',
  'yesterday_ds': '2020-12-01',
  'yesterday_ds_nodash': '20201201',
  'tomorrow_ds': '2020-12-03',
  'tomorrow_ds_nodash': '20201203',
  'END_DATE': '2020-12-02',
  'end_date': '2020-12-02',
  'dag_run': <DagRunparsing_recipes@2020-12-0209:56:05.289457+00:00:manual__2020-12-02T09:56:05.289457+00:00, externallytriggered: True>,
  'run_id': 'manual__2020-12-02T09:56:05.289457+00:00',
  'execution_date': <Pendulum[2020-12-02T09: 56: 05.289457+00:00]>,
  'prev_execution_date': datetime.datetime(2020,12,2,9,56,tzinfo=<TimezoneInfo[UTC,GMT,+00:00:00,STD]>),
  'next_execution_date': datetime.datetime(2020,12,2,9,58,tzinfo=<TimezoneInfo[UTC,GMT,+00:00:00,STD]>),
  'latest_date': '2020-12-02',
  'macros': <module'airflow.macros'from'/anaconda3/anaconda/lib/python3.6/site-packages/airflow/macros/__init__.py'>,
  'params': {},
  'tables': None,
  'task': <Task(PythonOperator): download_image>,
  'task_instance': <TaskInstance: parsing_recipes.download_image2020-12-02T09: 56: 05.289457+00:00[running]>,
  'ti': <TaskInstance: parsing_recipes.download_image2020-12-02T09: 56: 05.289457+00:00[running]>,
  'task_instance_key_str': 'parsing_recipes__download_image__20201202',
  'conf': <module'airflow.configuration'from'/anaconda3/anaconda/lib/python3.6/site-packages/airflow/configuration.py'>,
  'test_mode': False,
  'var': {
    'value': None,
    'json': None
  },
  'inlets': [],
  'outlets': [],
  'templates_dict': None
}

            
def download_image(**kwargs):
    local_image_file = None
    idx = 0
    records = []
    ti = kwargs['ti']

    parsed_records = ti.xcom_pull(key=None, task_ids='parse_recipes')

    for rec in parsed_records:
        idx += 1
        image_url = rec['image_url']
        r_url = rec['url']
        print('Downloading Pic# {}'.format(idx))
        local_image_file = dl_img(image_url, r_url)
        rec['local_image'] = local_image_file
        records.append(rec)

    return records

            
def store_data(**kwargs):
    ti = kwargs['ti']

    parsed_records = ti.xcom_pull(key=None, task_ids='download_image')
    print('PRINTING DUMPED RECORDS in STORE DATA')
    print(parsed_records)
    return 'store_data'

            
def store_data(**kwargs):
    ti = kwargs['ti']

    parsed_records = ti.xcom_pull(key=None, task_ids='download_image')
    connection = MySqlHook(mysql_conn_id='mysql_default')
    for r in parsed_records:
        url = r['url']
        data = json.dumps(r)
        sql = 'INSERT INTO recipes(url,data) VALUES (%s,%s)'
        connection.run(sql, autocommit=True, parameters=(url, data))
    return True

            
opr_email = EmailOperator(
        task_id='send_email',
        to='[email protected]',
        subject='Airflow Finished',
        html_content=""" <h3>DONE</h3> """,
        dag=dag
    )

            
[smtp]
# If you want airflow to send emails on retries, failure, and you want to use
# the airflow.utils.email.send_email_smtp function, you have to configure an
# smtp server here
smtp_host = smtp.gmail.com
smtp_starttls = False
smtp_ssl = True
# Uncomment and set the user/pass settings if you want to use SMTP AUTH
smtp_user = [email protected]
smtp_password = your_password
smtp_port = 465
smtp_mail_from = [email protected]

ETL Python Web Scraping

Introduzione ad Apache Airflow

scienzadeidati articoli - Introduzione ad AirflowI

            [2020-12-02 15:59:23,880] {__init__.py:51} INFO - Using executor SequentialExecutor
____________       _____________
____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ \_ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
_/_/  |_/_/  /_/    /_/    /_/  \____/____/|__/
v1.10.0
You will need to install mysqlclient as well to incorporate MySQL in your workflows. It is optional though.

            2020-12-02 22:36:24,943] {__init__.py:51} INFO - Using executor SequentialExecutor
/anaconda3/anaconda/lib/python3.6/site-packages/airflow/bin/cli.py:1595: 
DeprecationWarning: The celeryd_concurrency option in [celery] 
has been renamed to worker_concurrency - the old setting has been used,
but please update your config. default=conf.get('celery', 'worker_concurrency')),
____________       _____________
____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ \_ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
_/_/  |_/_/  /_/    /_/    /_/  \____/____/|__/
v1.10.0

            [2020-12-03 14:21:42,340] {__init__.py:57} INFO - Using executor SequentialExecutor
____________       _____________
____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ \_ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
_/_/  |_/_/  /_/    /_/    /_/  \____/____/|__/


/anaconda3/anaconda/lib/python3.6/site-packages/flask/exthook.py:71:
ExtDeprecationWarning: Importing flask.ext.cache is deprecated, use flask_cache instead.
.format(x=modname), ExtDeprecationWarning

[2020-12-03 14:21:43,119] [48995] {models.py:167} INFO - 
Filling up the DagBag from /Development/airflow_home/dags
Running the Gunicorn Server with:
Workers: 4 sync
Host: 0.0.0.0:8080

            import datetime as dt

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator


def greet():
    print('Writing in file')
    with open('path/to/file/greet.txt', 'a+', encoding='utf8') as f:
        now = dt.datetime.now()
        t = now.strftime("%Y-%m-%d %H:%M")
        f.write(str(t) + '\n')
    return 'Greeted'
def respond():
    return 'Greet Responded Again'

            with DAG('my_simple_dag',
         default_args=default_args,
         schedule_interval='*/10 * * * *',
         ) as dag:
    opr_hello = BashOperator(task_id='say_Hi',
                             bash_command='echo "Hi!!"')

    opr_greet = PythonOperator(task_id='greet',
                               python_callable=greet)
    opr_sleep = BashOperator(task_id='sleep_me',
                             bash_command='sleep 5')

    opr_respond = PythonOperator(task_id='respond',
                                 python_callable=respond)
                                 
opr_hello >> opr_greet >> opr_sleep >> opr_respond

            with DAG('my_simple_dag', 
                  catchup=False,
                  default_args=default_args,
                  schedule_interval='*/10 * * * *',
                  # schedule_interval=None,
                  ) as dag:

Pipeline Python

FastAPI con SQLAlchemy, PostgreSQL, Alembic e Docker – Parte 2 (versione asincrona)

scienzadeidati articoli - Database con FASTAPI [Parte2]

            # Pull base image
FROM python:3.7

# Set environment varibles
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

WORKDIR /code/

# Install dependencies
RUN pip install pipenv
COPY Pipfile Pipfile.lock /code/
RUN pipenv install --system --dev

COPY . /code/

EXPOSE 8000

            version: "3"

services:
  db:
    image: postgres:11
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=test_db
  web:
    build: .
    command: bash -c "uvicorn main:app --host 0.0.0.0 --port 8000 --reload"
    volumes:
      - .:/code
    ports:
      - "8000:8000"
    depends_on:
      - db

  pgadmin:
    container_name: pgadmin
    image: dpage/pgadmin4
    environment:
      - [email protected]
      - PGADMIN_DEFAULT_PASSWORD=admin
    ports:
      - "5050:80"
    depends_on:
      - db

            # db.py

import os
from databases import Database
from dotenv import load_dotenv
import sqlalchemy

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
load_dotenv(os.path.join(BASE_DIR, ".env"))

db = Database(os.environ["DATABASE_URL"])
metadata = sqlalchemy.MetaData()

            # app.py

from db import db
from fastapi import FastAPI


app = FastAPI(title="Async FastAPI")


@app.on_event("startup")
async def startup():
    await db.connect()


@app.on_event("shutdown")
async def shutdown():
    await db.disconnect()

            # model.py

from db import db

users = sqlalchemy.Table(
    "users",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("first_name", sqlalchemy.String),
    sqlalchemy.Column("last_name", sqlalchemy.String),
    sqlalchemy.Column("age", sqlalchemy.Integer),
)

            # model.py

import sqlalchemy
from db import db, metadata, sqlalchemy


users = sqlalchemy.Table(
    "users",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("first_name", sqlalchemy.String),
    sqlalchemy.Column("last_name", sqlalchemy.String),
    sqlalchemy.Column("age", sqlalchemy.Integer),
)


class User:
    @classmethod
    async def get(cls, id):
        query = users.select().where(users.c.id == id)
        user = await db.fetch_one(query)
        return user

    @classmethod
    async def create(cls, **user):
        query = users.insert().values(**user)
        user_id = await db.execute(query)
        return user_id

            # main.py

import uvicorn
from models import User as ModelUser
from schema import User as SchemaUser
from app import app
from db import db


@app.post("/user/")
async def create_user(user: SchemaUser):
    user_id = await ModelUser.create(**user.dict())
    return {"user_id": user_id}


@app.get("/user/{id}", response_model=SchemaUser)
async def get_user(id: int):
    user = await ModelUser.get(id)
    return SchemaUser(**user).dict()


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Database Docker FastAPI Python

FastAPI con SQLAlchemy, PostgreSQL e Alembic e ovviamente Docker – Parte 1

scienzadeidati articoli - Database con FASTAPI

            # main.py

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.post("/user/", response_model=User)
def create_user(user: User):
    return user

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

            # Dockerfile

# Pull base image
FROM python:3.7

# Set environment varibles
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

WORKDIR /code/

# Install dependencies
RUN pip install pipenv
COPY Pipfile Pipfile.lock /code/
RUN pipenv install --system --dev

COPY . /code/

EXPOSE 8000

            # docker-compose.yml

version: "3"

services:
  db:
    image: postgres:11
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=test_db
  web:
    build: .
    command: bash -c "alembic upgrade head && uvicorn main:app --host 0.0.0.0 --port 8000 --reload"
    volumes:
      - .:/code
    ports:
      - "8000:8000"
    depends_on:
      - db

  pgadmin:
    container_name: pgadmin
    image: dpage/pgadmin4
    environment:
      - [email protected]
      - PGADMIN_DEFAULT_PASSWORD=admin
    ports:
      - "5050:80"
    depends_on:
      - db

            ...
db:
    image: postgres:11
    ports:
        - "5432:5432"
    environment:
        - POSTGRES_USER=postgres
        - POSTGRES_PASSWORD=postgres
        - POSTGRES_DB=test_db

...

            from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = None

            from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import poolfrom alembic import context
# ---------------- added code here -------------------------#
import os, sys
from dotenv import load_dotenv

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
load_dotenv(os.path.join(BASE_DIR, ".env"))
sys.path.append(BASE_DIR)
#------------------------------------------------------------#
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# ---------------- added code here -------------------------#
# this will overwrite the ini-file sqlalchemy.url path
# with the path given in the config of the main code
config.set_main_option("sqlalchemy.url", os.environ["DATABASE_URL"])
#------------------------------------------------------------#
# Interpret the config file for Python logging.
# This line sets up loggers basically.
fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
# ---------------- added code here -------------------------#
import models
#------------------------------------------------------------#
# ---------------- changed code here -------------------------#
# here target_metadata was equal to None
target_metadata = models.Base.metadata
#------------------------------------------------------------#

            # models.py

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    first_name = Column(String,)
    last_name = Column(String)
    age = Column(Integer)

            ...
pgadmin:
    container_name: pgadmin
    image: dpage/pgadmin4
    environment:
      - [email protected]
      - PGADMIN_DEFAULT_PASSWORD=admin
    ports:
      - "5050:80"
    depends_on:
      - db
...

            # schema.py

from pydantic import BaseModel

class User(BaseModel):
    first_name: str
    last_name: str = None
    age: int
    class Config:
        orm_mode = True

            import uvicorn
from fastapi import FastAPI

#--------------- added code ------------------------#
import os
from fastapi_sqlalchemy import DBSessionMiddleware
from fastapi_sqlalchemy import db
from models import User as ModelUser
from schema import User as SchemaUser
from dotenv import load_dotenv

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
load_dotenv(os.path.join(BASE_DIR, ".env"))
#---------------------------------------------------#

app = FastAPI()
#--------------- added code ------------------------#
app.add_middleware(DBSessionMiddleware, 
                    db_url=os.environ["DATABASE_URL"])
#---------------------------------------------------#
#--------------- modified code ---------------------#
@app.post("/user/", response_model=SchemaUser)
def create_user(user: SchemaUser):
    db_user = ModelUser(first_name=user.first_name, 
                        last_name=user.last_name, 
                        age=user.age
    )
    db.session.add(db_user)
    db.session.commit()
    return db_user
#---------------------------------------------------#

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Database Docker FastAPI Python

Grafici delle Serie Temporali con Dash, Flask, TimescaleDB e Docker – Parte 3

scienzadeidati articoli - Grafici delle Serie Temporali con Dash, Flask, TimescaleDB, and Docker - Parte 3

            # /app/dash_setup.py

import dash
from flask.helpers import get_root_path


def register_dashapps(app):
    """
    Register Dash apps with the Flask app
    """

    # external Bootstrap CSS stylesheets
    external_stylesheets = [
        'https://cdn.jsdelivr.net/npm/[email protected]/dist/css/bootstrap.min.css'
    ]
    
    # external Bootstrap JavaScript files
    external_scripts = [
        "https://code.jquery.com/jquery-3.5.1.slim.min.js",
        "https://cdn.jsdelivr.net/npm/[email protected]/dist/umd/popper.min.js",
        "https://cdn.jsdelivr.net/npm/[email protected]/dist/js/bootstrap.min.js",
    ]

    # To ensure proper rendering and touch zooming for all devices, add the responsive viewport meta tag
    meta_viewport = [{
        "name": "viewport", 
        "content": "width=device-width, initial-scale=1, shrink-to-fit=no"
    }]

    dashapp = dash.Dash(
        __name__,
        # This is where the Flask app gets appointed as the server for the Dash app
        server = app,
        url_base_pathname = '/dash/',
        # Separate assets folder in "static_dash" (optional)
        assets_folder = get_root_path(__name__) + '/static_dash/', 
        meta_tags = meta_viewport, 
        external_scripts = external_scripts,
        external_stylesheets = external_stylesheets
    )
    dashapp.title = 'Dash Charts in Single-Page Application'

    # Some of these imports should be inside this function so that other Flask
    # stuff gets loaded first, since some of the below imports reference the other
    # Flask stuff, creating circular references 
    from app.dashapp.layout import get_layout
    from app.dashapp.callbacks import register_callbacks

    with app.app_context():

        # Assign the get_layout function without calling it yet
        dashapp.layout = get_layout

        # Register callbacks
        # Layout must be assigned above, before callbacks
        register_callbacks(dashapp)

    return None

            # /app/dashapp/layout.py

import os

from flask import url_for
import dash_html_components as html
import dash_core_components as dcc
import dash_bootstrap_components as dbc
from psycopg2.extras import RealDictCursor

# Local imports
from app.database import get_conn


def get_navbar():
    """Get a Bootstrap 4 navigation bar for our single-page application's HTML layout"""

    return dbc.NavbarSimple(
        children=[
            dbc.NavItem(dbc.NavLink("Blog", href="https://mccarthysean.dev")),
            dbc.NavItem(dbc.NavLink("IJACK", href="https://myijack.com")),
            dbc.DropdownMenu(
                children=[
                    dbc.DropdownMenuItem("References", header=True),
                    dbc.DropdownMenuItem("Dash", href="https://dash.plotly.com/"),
                    dbc.DropdownMenuItem("Dash Bootstrap Components", href="https://dash-bootstrap-components.opensource.faculty.ai/"),
                    dbc.DropdownMenuItem("Testdriven", href="https://testdriven.io/"),
                ],
                nav=True,
                in_navbar=True,
                label="Links",
            ),
        ],
        brand="Home",
        brand_href="/",
        color="dark",
        dark=True,
    )


def get_sensor_types():
    """Get a list of different types of sensors"""
    sql = """
        --Get the labels and underlying values for the dropdown menu "children"
        SELECT 
            distinct 
            type as label, 
            type as value
        FROM sensors;
    """
    conn = get_conn()
    with conn.cursor(cursor_factory=RealDictCursor) as cursor:
        cursor.execute(sql)
        # types is a list of dictionaries that looks like this, for example:
        # [{'label': 'a', 'value': 'a'}]
        types = cursor.fetchall()
    
    return types


def get_body():
    """Get the body of the layout for our Dash SPA"""

    types = get_sensor_types()

    # The layout starts with a Bootstrap row, containing a Bootstrap column
    return dbc.Row(
        [
            # 1st column and dropdown (NOT empty at first)
            dbc.Col(
                [
                    html.Label('Types of Sensors', style={'margin-top': '1.5em'}),
                    dcc.Dropdown(
                        options=types,
                        value=types[0]['value'],
                        id="types_dropdown"
                    )
                ], xs=12, sm=6, md=4
            ),
            # 2nd column and dropdown (empty at first)
            dbc.Col(
                [
                    html.Label('Locations of Sensors', style={'margin-top': '1.5em'}),
                    dcc.Dropdown(
                        # options=None,
                        # value=None,
                        id="locations_dropdown"
                    )
                ], xs=12, sm=6, md=4
            ),
            # 3rd column and dropdown (empty at first)
            dbc.Col(
                [
                    html.Label('Sensors', style={'margin-top': '1.5em'}),
                    dcc.Dropdown(
                        # options=None,
                        # value=None,
                        id="sensors_dropdown"
                    )
                ], xs=12, sm=6, md=4
            ),
        ]
    )


def get_layout():
    """Function to get Dash's "HTML" layout"""

    # A Bootstrap 4 container holds the rest of the layout
    return dbc.Container(
        [
            get_navbar(),
            get_body(), 
        ], 
    )

            # /app/dashapp/callbacks.py

import dash
import dash_core_components as dcc
import dash_html_components as html
from dash.dependencies import Input, Output, State
from psycopg2.extras import RealDictCursor

# Local imports
from app.database import get_conn


def get_sensor_locations(type_):
    """Get a list of different locations of sensors"""
    sql = f"""
        --Get the labels and underlying values for the dropdown menu "children"
        SELECT 
            distinct 
            location as label, 
            location as value
        FROM sensors
        WHERE type = '{type_}';
    """
    conn = get_conn()
    with conn.cursor(cursor_factory=RealDictCursor) as cursor:
        cursor.execute(sql)
        # locations is a list of dictionaries that looks like this, for example:
        # [{'label': 'floor', 'value': 'floor'}]
        locations = cursor.fetchall()
    
    return locations


def get_sensors(type_, location):
    """
    Get a list of sensor dictionaries from our TimescaleDB database, 
    along with lists of distinct sensor types and locations
    """
    sql = f"""
        --Get the labels and underlying values for the dropdown menu "children"
        SELECT 
            location || ' - ' || type as label,
            id as value
        FROM sensors
        WHERE 
            type = '{type_}'
            and location = '{location}';
    """
    conn = get_conn()
    with conn.cursor(cursor_factory=RealDictCursor) as cursor:
        cursor.execute(sql)
        # sensors is a list of dictionaries that looks like this, for example:
        # [{'label': 'floor - a', 'value': 1}]
        sensors = cursor.fetchall()

    return sensors


def register_callbacks(dash_app):
    """Register the callback functions for the Dash app, within the Flask app""" 

    @dash_app.callback(
        [
            Output("locations_dropdown", "options"),
            Output("locations_dropdown", "value")
        ],
        [
            Input("types_dropdown", "value")
        ]
    )
    def get_locations_from_types(types_dropdown_value):
        """Get the location options, based on the type of sensor chosen"""

        # First get the location options (i.e. a list of dictionaries)
        location_options = get_sensor_locations(types_dropdown_value)

        # Default to the first item in the list, 
        # and get the "value" from the dictionary
        location_value = location_options[0]["value"]

        return location_options, location_value


    @dash_app.callback(
        [
            Output("sensors_dropdown", "options"),
            Output("sensors_dropdown", "value")
        ],
        [
            Input("types_dropdown", "value"),
            Input("locations_dropdown", "value")
        ]
    )
    def get_sensors_from_locations_and_types(types_dropdown_value, locations_dropdown_value):
        """Get the sensors available, based on both the location and type of sensor chosen"""

        # First get the sensor options (i.e. a list of dictionaries)
        sensor_options = get_sensors(types_dropdown_value, locations_dropdown_value)

        # Default to the first item in the list, 
        # and get the "value" from the dictionary
        sensor_value = sensor_options[0]["value"]

        return sensor_options, sensor_value


    return None

            ...

def get_chart_row(): # NEW
    """Create a row and column for our Plotly/Dash time series chart"""

    return dbc.Row(
        dbc.Col(
            id="time_series_chart_col"
        )
    )


def get_layout():
    """Function to get Dash's "HTML" layout"""

    # A Bootstrap 4 container holds the rest of the layout
    return dbc.Container(
        [
            get_navbar(),
            get_body(), 
            get_chart_row(), # NEW
        ], 
    )

            def get_sensor_time_series_data(sensor_id):
    """Get the time series data in a Pandas DataFrame, for the sensor chosen in the dropdown"""

    sql = f"""
        SELECT 
            --Get the 3-hour average instead of every single data point
            time_bucket('03:00:00'::interval, time) as time,
            sensor_id,
            avg(temperature) as temperature,
            avg(cpu) as cpu
        FROM sensor_data
        WHERE sensor_id = {sensor_id}
        GROUP BY 
            time_bucket('03:00:00'::interval, time), 
            sensor_id
        ORDER BY 
            time_bucket('03:00:00'::interval, time), 
            sensor_id;
    """

    conn = get_conn()
    with conn.cursor(cursor_factory=RealDictCursor) as cursor:
        cursor.execute(sql)
        rows = cursor.fetchall()
        columns = [str.lower(x[0]) for x in cursor.description]

    df = pd.DataFrame(rows, columns=columns)
    
    return df

              @dash_app.callback(
        Output("time_series_chart_col", "children"),
        [Input("sensors_dropdown", "value")],
        [
            State("types_dropdown", "value"),
            State("locations_dropdown", "value")
        ]
    )
    def get_time_series_chart(
        sensors_dropdown_value, 
        types_dropdown_value, 
        locations_dropdown_value
    ):
        """Get the sensors available, based on both the location and type of sensor chosen"""

        df = get_sensor_time_series_data(sensors_dropdown_value)
        x = df["time"]
        y1 = df["temperature"]
        y2 = df["cpu"]
        title = f"Location: {locations_dropdown_value} - Type: {types_dropdown_value}"

        trace1 = go.Scatter(
            x=x,
            y=y1,
            name="Temp"
        )
        trace2 = go.Scatter(
            x=x,
            y=y2,
            name="CPU"
        )

        # Create two graphs using the traces above
        graph1 = get_graph(trace1, f"Temperature for {title}")
        graph2 = get_graph(trace2, f"CPU for {title}")

        return html.Div(
            [
                dbc.Row(dbc.Col(graph1)),
                dbc.Row(dbc.Col(graph2)),
            ]
        )

            def get_graph(trace, title):
    """Get a Plotly Graph object for Dash"""
    
    return dcc.Graph(
        # Disable the ModeBar with the Plotly logo and other buttons
        config=dict(
            displayModeBar=False
        ),
        figure=go.Figure(
            data=[trace],
            layout=go.Layout(
                title=title,
                plot_bgcolor="white",
                xaxis=dict(
                    autorange=True,
                    # Time-filtering buttons above chart
                    rangeselector=dict(
                        buttons=list([
                            dict(count=1,
                                label="1d",
                                step="day",
                                stepmode="backward"),
                            dict(count=7,
                                label="7d",
                                step="day",
                                stepmode="backward"),
                            dict(count=1,
                                label="1m",
                                step="month",
                                stepmode="backward"),
                            dict(step="all")
                        ])
                    ),
                    type = "date",
                    # Alternative time filter slider
                    rangeslider = dict(
                        visible = True
                    )
                )
            )
        )
    )

Docker Flask PGadmin4 Python TimeScaleDB

Dove archiviare enormi set di dati? Presentazione di ClickHouse

Come produrre un’elevata produttività? Presentazione di Kafka

Da dove ottenere i dati? Presentazione di CryptoFeed

Configurazione di Kafka e ClickHouse con Docker

Step 1: installazione di Docker e Docker Compose

Step 2: Creazione del Docker Compose per Clickhouse-Kafka

Preparazione dello script CryptoFeed

Step 1: scrittura dello script CryptoFeed

Step 2: Creazione del dockerfile di CryptoFeed

Configurazione di Clickhouse e Kafka

Step 1: creazione di tabelle MergeTree ClickHouse

Step 2: Creazione di tabelle nel motore Kafka

Step 3: Trasferire i dati tra tabelle con viste materializzate

Risoluzione dei problemi e test

Semplici controlli con ClickHouse

Controllo dei broker e degli argomenti di Kafka

Controllo dei log di CryptoFeed

Conclusione

Il progetto

Il setup

Parte I: Il Web Scraper

Dove?

Cosa?

Come?

Trovare il selettore CSS di un elemento.

Il file ‘scrape.json’

Codificare lo scraping.

Parte II: L’API REST

Configurazione FastAPI

Definizione di un endpoint

Filtraggio

Conclusione

Riferimenti

Requisiti

Come funziona Celery

Obiettivi

Inizializzazione

Perché Celery e RabbitMQ? Perché non un'altra tecnologia?

Configurazione di RabbitMQ

Testare Celery con RabbitMQ

Spiegazione

Terminale #2

Terminale #3:

Terminale 2:

Creazione del task.py con Celery per il web scraping

Miglioramenti alle funzioni di scraping

Pianificazione dei task con Celery:

Creazione dello scheduler

Esecuzione dei task

Terminale #1

Terminale #2

Conclusione

Quali sono i prossimi passi?

Ulteriori letture

Requisiti

Obiettivi

Inizializzazione

Importazione delle librerie

Testare la richiesta

Scraping del contenuto

Analisi dei dati

Output in un file

Conclusione

Introduzione ai microservice

Vantaggi dei microservice

Svantaggi dei microservice

Perché Microservice in Python?

Introduzione a FastAPI

Perché FastAPI

Installazione FastAPI

Creazione di un semplice REST API utilizzando FastAPI

Utilizzo del database PostgreSQL con FastAPI

Modelli di gestione dei dati nei microservice

Database per servizio

Database condiviso

Composizione API

Creazione di un microservice Python in Docker

Installazione di Docker e Docker Compose

Creazione del servizio Movies

Creazione del casts-service