https://studystepbystep.notion.site/Middle-Python-Developer-c111da8f3ce14b3eaa12d6570d5c5ae0
Готовое тестовое задание отправляйте Карине Аруштовой: [email protected]
Тестовое задание на Middle Python Developer
Ответы можно оформить в любой удобном формате: Notion-страница, Google Docs и тд.
Вы можете сделать любые два задания.
Задания
Task: get new records from DataBase1 with Notion (foreign key) and write to DataBase2.Notion (key).
Задача 1: Работа с PostgreSQL/SQL
Цель: Оценить навыки работы с базами данных и понимание SQL.
Задание: Написать запрос SQL, который получает все обновления из DataBase1 Notion и записывает в БД SQL, где:
DataBase1 содержит данные:
Дата - формат Date
Компания - Relation на таблицу DataBase2
Пользователь - User Notion (по нему получаем ID и почту)
Наименование - индексное поле Notion
DataBase2 содержит:
Наименование - индексное поле Notion
Адрес - Text
DROP TABLE DataBase1;
DROP TABLE DataBase2;
DROP TABLE BdSQL;
CREATE TABLE DataBase1 (
Date INTEGER, -- epoch
Company INTEGER,
User_Notion INTEGER,
Notion INTEGER, -- foreign key
FOREIGN KEY(Notion) REFERENCES DataBase2(Notion)
);
CREATE TABLE DataBase2 (
Notion INTEGER PRIMARY KEY,
Address TEXT
);
CREATE TABLE BDSQL (
Idfk INTEGER,
FOREIGN KEY(Idfk) REFERENCES Database2(Id)
);
INSERT INTO DataBase1 VALUES(2147483640, 2, 1, 1);
INSERT INTO DataBase1 VALUES(2147483641, 2, 1, 1);
INSERT INTO DataBase1 VALUES(2147483642, 2, 1, 1);
INSERT INTO DataBase2 VALUES(1, 'Street1');
# -- Create unique key in DataBase1
ALTER TABLE DataBase1 ADD COLUMN Id INTEGER;
UPDATE DataBase1
SET Id = (
SELECT COUNT(*)
FROM DataBase1 AS t
WHERE t.rowid <= DataBase1.rowid
);
# -- update BDSQL with new references to DataBase1
INSERT INTO BDSQL
SELECT Id
FROM DataBase1 AS t
WHERE t.Date >= 2147483641;
# -- test
SELECT * FROM DataBase1;
SELECT Null;
SELECT * FROM DataBase2;
SELECT Null;
SELECT * FROM BDSQL;
Date | Company | User_Notion | Notion | Id |
---|---|---|---|---|
2147483640 | 2 | 1 | 1 | 1 |
2147483641 | 2 | 1 | 1 | 2 |
2147483642 | 2 | 1 | 1 | 3 |
Null | ||||
Notion | Address | |||
1 | Street1 | |||
Null | ||||
Idfk | ||||
2 | ||||
3 |
Задача 2: Разработка Telegram бота
Задание: Создать базовую версию Telegram бота, который может
регистрировать новых пользователей, сохраняя их в базу данных, и
отвечать на простые команды, например, выводить справочную информацию
или статистику пользователя.
- MTProto Mobile Protocol - access to a server API from applications running on mobile devices. ( mobile telegram client use it)
- Bot API - HTTP API https://core.telegram.org/bots/api - GET/POST https://api.telegram.org/.........
- bot development frameworks: Telegraf (Node.js), python-telegram-bot (Python), and BotMan (PHP)
- GUI platforms
- Bot API - own instance https://github.com/tdlib/telegram-bot-api
- Max File Upload: 2G
- Telegram APIs https://core.telegram.org/methods
- insist to register your application in our system.
- TDLib (Telegram Database Library) - customized Telegram clients.
- github https://github.com/tdlib/td
- https://core.telegram.org/
- doc https://core.telegram.org/tdlib
- each TDLib instance handles more than 24000 active bots simultaneously.
- methods https://core.telegram.org/methods
https://core.telegram.org/tdlib/getting-started
Have C++, Java, and .NET interfaces. for most use cases we suggest to use the JSON interface.
Async steps:
- ClientManager.send
- ClientManager.receive
[ dont-want-to-share-for-free ]
JSON interface. https://github.com/tdlib/td#using-json
- main documentation https://core.telegram.org/tdlib/docs/td__json__client_8h.html
- main source https://github.com/tdlib/td/blob/master/td/telegram/td_json_client.h
- TDLib methods https://core.telegram.org/tdlib/docs/classtd_1_1td__api_1_1_function.html
- TDLib classes https://core.telegram.org/tdlib/docs/classtd_1_1td__api_1_1_object.html
- TDLib updates https://core.telegram.org/tdlib/docs/classtd_1_1td__api_1_1_update.html
- references https://core.telegram.org/tdlib/docs/index.html#using-json
basics:
- API objects are represented as JSON objects with the same keys as the API object field names.
- “@type” - type name, (optional where type is uniquely determined by the context)
- Boolean - as Bool
- int32, int53, double - Number
- int64 and string - as String.
- base64 - encoded and stored as String
- array - as Array
- “@extra” - field can be added to request to find it in response
- “@client_id” - identifier of the client for which a response or an update was received.
main functions:
- int td_create_client_id ()
void td_send (int client_id, const char *request)
[ dont-want-to-share-for-free ]
[ dont-want-to-share-for-free ]
examples https://github.com/tdlib/td/tree/master/example#readme
- 3m official https://github.com/tdlib/td/blob/master/example/python/tdjson_example.py
- 2w 51s users/bots https://github.com/pytdbot/client
- uses JSON libs for speed:
- https://github.com/ijl/orjson#install
- https://github.com/ultrajson/ultrajson#ultrajson
- 2m 94s high-level API https://github.com/pylakey/aiotdlib
- 2y https://github.com/alvhix/pywtdlib
- 4y https://github.com/JunaidBabu/tdlib-python
steps:
- load shared library - tdjson
- load TDLib functions from shared library - tdjson. …
- initialize TDLib log with desired parameters - _td_set_log_message_callback, define td_execute
- setting TDLib log verbosity level to 1 (errors) - with td_execute
- create client call - _td_create_client_id()
- define wrappers for - td_send(query), td_receive()
from ctypes.util import find_library
from ctypes import *
import json
import os
import sys
# 1) --- --- load shared library
tdjson_path = find_library('tdjson')
if tdjson_path is None:
if os.name == 'nt':
tdjson_path = os.path.join(os.path.dirname(__file__), 'tdjson.dll')
else:
sys.exit("Can't find 'tdjson' library")
tdjson = CDLL(tdjson_path)
# 2) --- --- load TDLib functions from shared library
_td_create_client_id = tdjson.td_create_client_id
_td_create_client_id.restype = c_int
_td_create_client_id.argtypes = []
_td_receive = tdjson.td_receive
_td_receive.restype = c_char_p
_td_receive.argtypes = [c_double]
_td_send = tdjson.td_send
_td_send.restype = None
_td_send.argtypes = [c_int, c_char_p]
_td_execute = tdjson.td_execute
_td_execute.restype = c_char_p
_td_execute.argtypes = [c_char_p]
log_message_callback_type = CFUNCTYPE(None, c_int, c_char_p)
_td_set_log_message_callback = tdjson.td_set_log_message_callback
_td_set_log_message_callback.restype = None
_td_set_log_message_callback.argtypes = [c_int, log_message_callback_type]
# 3) --- --- initialize TDLib log with desired parameters
@log_message_callback_type
def on_log_message_callback(verbosity_level, message):
if verbosity_level == 0:
sys.exit('TDLib fatal error: %r' % message)
def td_execute(query):
query = json.dumps(query).encode('utf-8')
result = _td_execute(query)
if result:
result = json.loads(result.decode('utf-8'))
return result
_td_set_log_message_callback(2, on_log_message_callback)
# 4) --- --- setting TDLib log verbosity level to 1 (errors)
print(str(td_execute({'@type': 'setLogVerbosityLevel', 'new_verbosity_level': 1, '@extra': 1.01234})).encode('utf-8'))
# 5) --- --- create client
client_id = _td_create_client_id()
# 6) --- --- simple wrappers for client usage
def td_send(query):
query = json.dumps(query).encode('utf-8')
_td_send(client_id, query)
def td_receive():
result = _td_receive(1.0)
if result:
result = json.loads(result.decode('utf-8'))
return result
# 7) --- --- another test for TDLib execute method
print(str(td_execute({'@type': 'getTextEntities', 'text': '@telegram /test_command https://telegram.org telegram.me', '@extra': ['5', 7.0, 'a']})).encode('utf-8'))
# start the client by sending a request to it
td_send({'@type': 'getOption', 'name': 'version', '@extra': 1.01234})
# main events cycle
while True:
event = td_receive()
if event:
# process authorization states
if event['@type'] == 'updateAuthorizationState':
auth_state = event['authorization_state']
# if client is closed, we need to destroy it and create new client
if auth_state['@type'] == 'authorizationStateClosed':
break
# set TDLib parameters
# you MUST obtain your own api_id and api_hash at https://my.telegram.org
# and use them in the setTdlibParameters call
if auth_state['@type'] == 'authorizationStateWaitTdlibParameters':
td_send({'@type': 'setTdlibParameters',
'database_directory': 'tdlib',
'use_message_database': True,
'use_secret_chats': True,
'api_id': 94575,
'api_hash': 'a3406de8d171bb422bb6ddf3bbd800e2',
'system_language_code': 'en',
'device_model': 'Desktop',
'application_version': '1.0'})
# enter phone number to log in
if auth_state['@type'] == 'authorizationStateWaitPhoneNumber':
phone_number = input('Please enter your phone number: ')
td_send({'@type': 'setAuthenticationPhoneNumber', 'phone_number': phone_number})
# enter email address to log in
if auth_state['@type'] == 'authorizationStateWaitEmailAddress':
email_address = input('Please enter your email address: ')
td_send({'@type': 'setAuthenticationEmailAddress', 'email_address': email_address})
# wait for email authorization code
if auth_state['@type'] == 'authorizationStateWaitEmailCode':
code = input('Please enter the email authentication code you received: ')
td_send({'@type': 'checkAuthenticationEmailCode',
'code': {'@type': 'emailAddressAuthenticationCode', 'code' : code}})
# wait for authorization code
if auth_state['@type'] == 'authorizationStateWaitCode':
code = input('Please enter the authentication code you received: ')
td_send({'@type': 'checkAuthenticationCode', 'code': code})
# wait for first and last name for new users
if auth_state['@type'] == 'authorizationStateWaitRegistration':
first_name = input('Please enter your first name: ')
last_name = input('Please enter your last name: ')
td_send({'@type': 'registerUser', 'first_name': first_name, 'last_name': last_name})
# wait for password if present
if auth_state['@type'] == 'authorizationStateWaitPassword':
password = input('Please enter your password: ')
td_send({'@type': 'checkAuthenticationPassword', 'password': password})
# handle an incoming update or an answer to a previously sent request
print(str(event).encode('utf-8'))
sys.stdout.flush()
[ dont-want-to-share-for-free ]
[ dont-want-to-share-for-free ]
[ dont-want-to-share-for-free ]
[ dont-want-to-share-for-free ]
- build, exampels, links https://core.telegram.org/tdlib/docs/
- https://core.telegram.org/tdlib/options
- https://core.telegram.org/tdlib/notification-api/
ways
- USER: standard login code flow https://core.telegram.org/api/auth
- BOT: token generated by @botfather. message @BotFather on Telegram to register your bot and receive its authentication token. -
- simplified HTTP Bot API
pros: All messages from channels where they are a member.
cons:
- In Groups they are limited by see only relevant messages, except bots that was added as admins.
- channels
In group without admin rights :
- only see commands /command@this_bot
- can send messages
- flooding
- spamming
- faking subscriber
- view counters of channels
- generate token at @BotFather
- register and get api_id and api_hash at https://my.telegram.org/apps
- /setjoingroups or https://telegram.me/BOT_USER_NAME?startgroup=true
- add bot to group M-x telega-chat-add-member
[ dont-want-to-share-for-free ]
Задача 3: Документирование и устранение ошибок
Цель: Проверить навыки написания документации и способность к анализу и устранению ошибок в коде.
Задание:
import sqlite3
def connect_to_db(db_name):
try:
conn = sqlite3.connect(db_name)
return conn
except Exception as e:
print(f"An error occurred: {e}")
def create_table(conn):
cursor = conn.cursor()
cursor.execute("""CREATE TEBLE users(
id INTEGER PRIMARY KAY AUTOINCREMENT,
name TEXT NOT NULL,
age INTEGER)""")
conn.commit()
def insert_user(conn, user_name, user_age):
cursor = conn.cursor()
cursor.execute("INSER INTO users (name, age) VALUES (?, ?)", (user_name, user_age))
conn.commit()
# Usage example
db_connection = connect_to_db('my_database.db')
create_table(db_connection)
insert_user(db_connection, 'Alice', 30)
Answer:
1. The table creation query has a typo in "CREATE TEBLE users", it should be "CREATE TABLE
users".
2. There is a typo in "id INTEGER PRIMARY KAY AUTOINCREMENT", it should be "id INTEGER PRIMARY
KEY AUTOINCREMENT".
3. In the insert user function, there is a typo in "INSER INTO users", it should be "INSERT INTO
users".
4. The data type for the "age" column is missing in the table creation query. It should be "age
INTEGER".
5. The column names in the insert query and the values provided do not match. The table has
"name" and "age" columns, but the query is inserting into "fullname" and "user_age".
These corrections should be made to ensure the code works as intended.
Task: write documentation and correct mistakes and bugs, demonstate documentation skills.
Plan:
- run code
- fix errors
- add comments
- linting
- write report about found errors and all changes
Final result:
import sqlite3
print(sqlite3.version)
2.6.0
Solution:
""" This program is a task from "Step By Step" copmany for job Machine
Learning Engineer. This is modified version with fixed errors, added
comments and type hints.
"""
import sqlite3
from sqlite3 import Connection
import sys
def connect_to_db(db_name) -> None | Connection:
"Create connection to a database file in current directory."
try:
conn = sqlite3.connect(db_name)
return conn
except Exception as e:
print(f"An error occurred: {e}")
return None
def create_table(conn: Connection):
"Create table users."
cursor = conn.cursor()
cursor.execute("""CREATE TABLE users(
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
age INTEGER)""")
conn.commit()
def insert_user(conn: Connection, user_name: str, user_age: int):
"Insert row to users table."
cursor = conn.cursor()
cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)",
(user_name, user_age))
conn.commit()
def creat_and_fill_users(db_name: str):
"Create table users and insert one row."
# Usage example
db_connection = connect_to_db(db_name)
if db_connection is None:
sys.exit(1)
create_table(db_connection)
insert_user(db_connection, 'Alice', 30)
def test_drop(db_name: str, table_name: str):
"Drop table if exist."
db_connection = connect_to_db(db_name)
cur = db_connection.cursor()
cur.execute("DROP TABLE IF EXISTS " + table_name)
def test_select(db_name: str, table_name: str):
"Select and print all rows from table."
db_connection = connect_to_db(db_name)
cur = db_connection.cursor()
cur.execute("SELECT * FROM " + table_name)
rows = cur.fetchall()
for row in rows:
print(row)
if __name__ == "__main__":
DB_NAME_ARG = 'my_database.db'
test_drop(DB_NAME_ARG, 'users')
creat_and_fill_users(DB_NAME_ARG)
test_select(DB_NAME_ARG, 'users')
(1, 'Alice', 30)
- Found three typos:
- 1. The table creation query has a typo in “CREATE TEBLE users”,
it should be “CREATE TABLE users”.
- 2. There is a typo in “id INTEGER PRIMARY KAY AUTOINCREMENT”, it
should be “id INTEGER PRIMARY KEY AUTOINCREMENT”.
- 1. In the insert user function, there is a typo in “INSER INTO
users”, it should be “INSERT INTO users”.
- For purpose of testing we wrap code in a function “create_table” and “test_select” and “test_drop”
- From original code we only remove hardcoded database name for connection, everythin else was leave untouched, beacause our purpose is only fix error, not to make refactoring.
- Linting: PyLint give us 2 warnings: too general exception and ” Either all return statements in a function should return an expression, or none of them should.” for connect_to_db function. To fix that we halt program if None is returned, for second error we return None at the end of the function. pycodestyle: In insert_user line was too long.