📶 🍋 ♦️ विकिपीडिया से अप-टू-डेट COVID-19 डेटा के साथ पायथन फ्लास्क सेवा बनाएं 🔂 🧑🏽 🤸

हमारी टीम कई देशों, शहरों और क्षेत्रों के लिए वैश्विक डेटा प्रदर्शित करने के लिए एक सूचना सेवा का निर्माण कर रही है । इस वर्ष के फरवरी के अंत तक, दुनिया भर में कोरोनावायरस के तेजी से प्रसार ने हमें अपने आवेदन में स्थिति की निगरानी के लिए अतिरिक्त कार्यक्षमता शुरू करने के लिए प्रेरित किया। वेब इंटरफ़ेस में डेटा की कल्पना करने के अलावा, इस कार्य का मुख्य घटक लोकप्रिय फ्लास्क वेब फ्रेमवर्क का उपयोग करके पायथन में लिखा गया एक माइक्रोसेवा था।

सेवा नियमित रूप से विभिन्न स्रोतों से डेटा अपडेट करती है, और अनुरोध पर, वेब इंटरफ़ेस में विज़ुअलाइज़ेशन के लिए आवश्यक जानकारी प्रदान करती है। मुख्य डेटा स्रोत देशों और क्षेत्रों में वायरस के प्रसार पर विकिपीडिया पृष्ठ हैं। इन पृष्ठों पर संकेतक के साथ तालिकाओं को जल्दी से अपडेट किया जाता है और संक्रमण के प्रसार की निगरानी के लिए सेवा के डेटा स्रोत के रूप में उत्कृष्ट हैं।

लेख में, मैं ग्राहक के अनुरोधों के लिए एपीआई बनाने के लिए डेटा प्राप्त करने और अद्यतन करने से, सेवा के मुख्य घटकों के बारे में बात करूंगा। प्रोजेक्ट कोड github रिपॉजिटरी में उपलब्ध है ।

, Python Flask . , , . , COVID-19.

Flask

Python , ORM Python SQLAlchemy. Routitude PostgreSQL.

pip install requirements.txt

URI :

export COVID19API_DB_URI=<  URI, : postgresql://localhost/covid19api>

. :

/api                     API              
    covid.py            API c   COVID-19  
/datasources           
    /test                 
        test_covid.py       COVID-19  
    covid_wiki.py         COVID-19    
    utils.py               
/mirgations               SQLAlchemy
    ...                 Flask-Migrate (Alembic)
/test                  
    test_app.py         (API)
app.py                      HTTP 
appvars.py               
config.py              
manage.py              CLI     
models.py             ORM       
requirements.txt

, :

, ;
ORM COVID-19;
;
API , .

. github , , .

. :

html read_html Pandas. . . , , . , , . , . Pandas DataFrame, , html .

datasources/utils.py

import pandas as pd

def get_wiki_table_df(page_url, match_string):
    response = requests.get(page_url)
    tables = pd.read_html(response.content)
    df = None
    for table in tables:
        df = table
        if match_string in str(df):
            break
    return df

. . , COVID-19 .

datasources/covid_wiki.py

def get_report_countries():
    url = (
        'https://en.wikipedia.org/wiki/'
        '2019%E2%80%9320_coronavirus_pandemic_by_country_and_territory'
    ) 
    df = utils.get_wiki_table_df(url, 'Locations[b]')
    df = pd.DataFrame(
        df.values[:, 1:5], 
        columns=['country', 'confirmed', 'deaths', 'recovered']
    )
    df = df[~df['country'].isna()]
    df['country'] = df['country'].apply(lambda x: utils.clean_territory_name(x))
    df.drop(df[df['country'].str.len() > 40].index, inplace=True)
    df = utils.wiki_table_df_numeric_column_clean(df, [
        'confirmed', 'deaths', 'recovered'
    ])
    df['state'] = None
    check_report(df)
    return df

html , , , .

datasources/test/test_covid.py

from unittest import TestCase
from datasources import covid_wiki

class TestCovid(TestCase):
    def test_get_wiki_report(self):
        report = covid_wiki.get_report_countries()
        self.assertTrue('Russia' in list(report['country']))
        self.assertTrue(report.shape[0] > 0)

nosetests datasources

SQLAlchemy ORM Flask

, , . Python , , , . - (Object-Relational Mapping, ORM). Python ORM SQLAlchemy. , , , SQLAchemy . Alembic. , Flask , Flask-Migrate. appvars.py manage.py.

ORM Model SQLAlchemy. SQLAchemy . , . ORM .

models.py

class CovidWiki(db.Model):
    __tablename__ = 'covid_wiki'

    territory_id = Column(
        db.VARCHAR(length=256), nullable=False, primary_key=True
    )
    update_time = Column(db.TIMESTAMP(), nullable=False)
    country = Column(db.VARCHAR(length=128), nullable=False)
    state = Column(db.VARCHAR(length=128), nullable=True)
    confirmed = Column(db.INTEGER(), nullable=True)
    deaths = Column(db.INTEGER(), nullable=True)
    recovered = Column(db.INTEGER(), nullable=True)

Index('ix_covid_wiki_country', CovidWiki.country)
Index('ix_covid_wiki_state', CovidWiki.state)

, , Alembic:

python manage.py db init

migrations . ORM :

python manage.py db migrate -m covid_wiki

, , , , , , . migrations/versions. , , python, . 2 — upgrate downgrade. ORM , , .

, . :

python manage.py db upgrade

COVID-19

ORM , pandas DataFrame , , , . , . , , SQLAlchemy.

models.py

def update_data_by_dataframe(self, df):
    report = df.to_dict(orient='records')
    report_last = self.get_wiki_last_report()
    for value in report:
        territory_id = self.get_id(value['country'], value['state'])
        value['territory_id'] = territory_id
        changed = (
            (len(report_last) == 0) or
            (territory_id not in report_last) or
            (utils.get_covid_values_sum(value) !=
             utils.get_covid_values_sum(report_last[territory_id]))
        )
        if not changed:
            continue
        logging.info(f"Updating data for territory: {territory_id}")
        data = dict(value)
        for name in utils.STAT_NAMES:
            value = data[name]
            if np.isnan(value):
                data[name] = None
                continue
            data[name] = int(value)
        data['update_time'] = datetime.datetime.now()
        report = CovidWiki(**data)
        db.session.merge(report)
        db.session.commit()

. .

models.py

def update_data(self):
    logging.info('Updating countries data')
    self.update_data_by_dataframe(covid_wiki.get_report_countries())
    logging.info('Updating Russian states data')
    self.update_data_by_dataframe(covid_wiki.get_report_ru())
    logging.info('Updating USA states data')
    self.update_data_by_dataframe(covid_wiki.get_report_us())

Flask Flask-Script . , , manager.command . .

manage.py

@manager.command
def update_covid_data():
    CovidWiki().update_data()

, , :

python manage.py update_covid_data

- , cron.

API COVID-19 Flask

. Flask . , SQLAlchemy . , COVID-19:

;
;
.

api/covid.py

def get_covid_countries_report():
    data = db.session.query(CovidWiki).filter(CovidWiki.state.is_(None)).all()
    return [v.to_dict() for v in data]

def get_covid_states_report_by_country(country):
    data = db.session.query(CovidWiki).filter(and_(
        CovidWiki.state.isnot(None),
        func.lower(CovidWiki.country) == country.lower(),
    )).all()
    return [v.to_dict() for v in data]

def get_covid_total_stats():
    def to_dict(v):
        return {'confirmed': v[0], 'deaths': v[1], 'recovered': v[2]}

    curr = db.session.query(
        func.sum(CovidWiki.confirmed),
        func.sum(CovidWiki.deaths),
        func.sum(CovidWiki.recovered),
        func.max(CovidWiki.update_time)
    ).filter(CovidWiki.state.is_(None)).one()
    return {
        'data': to_dict(curr),
        'last_update_time': utils.datetime2string(curr[3], time=True)
    }

API. , , .

app.py

@app.route('/covid/countries')
def get_covid_countries_report():
    report = covid_api.get_covid_countries_report()
    check_data(report)
    return jsonify(report)

@app.route('/covid/states/<string:country>')
def get_covid_states_report_by_country(country):
    report = covid_api.get_covid_states_report_by_country(country)
    check_data(report)
    return jsonify(report)

@app.route('/covid/total')
def get_covid_total_stats():
    report = covid_api.get_covid_total_stats()
    check_data(report)
    return jsonify(report)

Flask .

python app.py

, , , curl.

curl http://localhost:5000/covid/total

. test, :

nosetests

-. Python gunicorn uWSGI.

, . API , . , .

पर Routitude, हम नक्शा और डैशबोर्ड पर COVID -19 संक्रमण की वर्तमान दरों को प्रदर्शित करने के लिए इस सेवा का उपयोग करें। सभी स्रोत सामग्री गितुब भंडार में उपलब्ध हैं । किसी भी सुधार, बग फिक्स, नई सुविधाओं और डेटा का स्वागत है। परियोजना में सुधार के लिए लेख और सुझावों पर किसी भी टिप्पणी से मुझे खुशी होगी।

विकिपीडिया से अप-टू-डेट COVID-19 डेटा के साथ पायथन फ्लास्क सेवा बनाएं

Flask

SQLAlchemy ORM Flask

COVID-19

API COVID-19 Flask

More articles: