HW 6 (Kudinov Nikita)

homeworks/relby/6/csv_request_xml/__main__.py

@@ -0,0 +1,50 @@
+import asyncio
+from typing import Final
+
+from loguru import logger
+
+from csv_request_xml.api_types import Email, User
+from csv_request_xml.reader import read_emails_from_csv
+from csv_request_xml.requester import get_user_attributes, get_users
+from csv_request_xml.saver import save_to_xml
+
+EMAILS_FILEPATH: Final = 'emails.csv'
+
+
+def _configure_logger():
+    logger.add(lambda _: exit(1), level='ERROR')  # noqa: WPS421
+
+
+async def handle_email(email: Email, users: list[User]):
+    logger.info('Start working with `{0}`'.format(email))
+    user = next(
+        (user for user in users if user['email'] == email),
+        None,
+    )
+    if user is not None:
+        attrs = await get_user_attributes(user)
+        save_to_xml(user, attrs)
+    else:
+        logger.warning(
+            'User with this email `{0}` was not found'.format(email),
+        )
+
+
+async def async_main() -> None:
+    _configure_logger()
+    emails = read_emails_from_csv(EMAILS_FILEPATH)
+    users = await get_users()
+
+    await asyncio.gather(*[
+        handle_email(email, users)
+        for email in emails
+    ])
+
+
+def main() -> None:
+    asyncio.run(async_main())
+
+
+if __name__ == '__main__':
+    main()
+# noqa: WPS473 (I suppose it's a bug)

homeworks/relby/6/csv_request_xml/api_types.py

@@ -0,0 +1,42 @@
+import types
+from typing import Final, TypeAlias, TypedDict
+
+Email: TypeAlias = str
+
+USERS_API_URL: Final = 'https://jsonplaceholder.typicode.com/users'
+
+
+# Response types
+class User(TypedDict):
+    id: int
+    email: str
+
+
+class Post(TypedDict):
+    id: int
+    title: str
+    body: str
+
+
+class Album(TypedDict):
+    id: int
+    title: str
+
+
+class Todo(TypedDict):
+    id: int
+    title: str
+    completed: bool
+
+
+class UserAttrs(TypedDict):
+    posts: list[Post]
+    albums: list[Album]
+    todos: list[Todo]
+
+
+BINDINGS: Final = types.MappingProxyType({
+    'posts': Post,
+    'albums': Album,
+    'todos': Todo,
+})

homeworks/relby/6/csv_request_xml/reader.py

@@ -0,0 +1,45 @@
+import csv
+import os
+import re
+
+from loguru import logger
+
+from csv_request_xml.api_types import Email
+
+
+def _is_email(entry: str) -> bool:
+    regex = re.compile(
+        r'([A-Za-z0-9]+[.-_])*[A-Za-z0-9]+@[A-Za-z0-9-]+(\.[A-Z|a-z]{2,})+',
+    )
+    return re.match(regex, entry) is not None
+
+
+def read_emails_from_csv(filepath: str) -> list[Email]:
+    if not os.path.exists(filepath):
+        logger.error('csv file with filepath `{0}` was not found'.format(
+            filepath,
+        ))
+
+    logger.info('Start reading {0}'.format(filepath))
+    emails: list[Email] = []
+    with open(filepath, 'r') as csv_file:
+        reader = csv.reader(csv_file)
+        for line in reader:
+            for entry in line:
+                if _is_email(entry):
+                    emails.append(entry)
+                else:
+                    logger.warning(
+                        '`{0}` is not valid email. Skipping it.'.format(
+                            entry,
+                        ),
+                    )
+    if emails:
+        logger.success('{0} emails were read from csv file.'.format(
+            len(emails),
+        ))
+    else:
+        logger.error('No valid username emails were found in {0}.'.format(
+            filepath,
+        ))
+    return emails

homeworks/relby/6/csv_request_xml/requester.py

@@ -0,0 +1,61 @@
+import asyncio
+from typing import cast
+
+import httpx
+from loguru import logger
+
+from csv_request_xml.api_types import BINDINGS, USERS_API_URL, User, UserAttrs
+
+
+def _only_keys(dictionary: dict, keys: list) -> dict:
+    return {
+        dict_value: dictionary[dict_value]
+        for dict_value in dictionary
+        if dict_value in keys
+    }
+
+
+async def get_users() -> list[User]:
+    async with httpx.AsyncClient() as client:
+        request = await client.get(USERS_API_URL)
+        logger.info('Request to {0} took {1} '.format(
+            USERS_API_URL,
+            request.elapsed,
+        ))
+        return request.json()
+
+
+async def get_user_attribute(
+    user: User,
+    attr_name: str,
+    client: httpx.AsyncClient,
+) -> list[dict]:
+    related_type = BINDINGS[attr_name]
+    api_route = '{0}/{1}/{2}'.format(
+        USERS_API_URL,
+        user['id'],
+        attr_name,
+    )
+    response = await client.get(api_route)
+    logger.info('Request to {0} took {1}'.format(
+        api_route,
+        response.elapsed,
+    ))
+    return [
+        _only_keys(attr, list(related_type.__annotations__.keys()))
+        for attr in response.json()
+    ]
+
+
+async def get_user_attributes(user: User) -> UserAttrs:  # noqa: WPS210
+    out = {}
+    async with httpx.AsyncClient() as client:
+        attr_names = UserAttrs.__annotations__
+        user_attributes = await asyncio.gather(*[
+            get_user_attribute(user, attr_name, client)
+            for attr_name in attr_names
+        ])
+        for attr_name, user_attribute in zip(attr_names, user_attributes):
+            out[attr_name] = user_attribute
+
+    return cast(UserAttrs, out)

homeworks/relby/6/csv_request_xml/saver.py

@@ -0,0 +1,47 @@
+import os
+
+from loguru import logger
+from lxml import etree  # noqa: S410
+
+from csv_request_xml.api_types import User, UserAttrs
+
+
+def save_to_xml(
+    user: User,
+    attrs: UserAttrs,
+    filepath: str = 'users',
+) -> None:
+    os.makedirs(filepath, exist_ok=True)
+
+    user_id = user['id']
+
+    user_node = etree.Element('user')
+
+    etree.SubElement(user_node, 'id').text = str(user_id)
+    etree.SubElement(user_node, 'email').text = user['email']
+
+    _add_attr_to_xml(attrs, user_node)
+
+    tree = etree.ElementTree(user_node)
+    tree.write('{0}/{1}.xml'.format(filepath, user_id), pretty_print=True)
+    logger.success('Saved {0}/{1}.xml for user with email `{2}`'.format(
+        filepath,
+        user_id,
+        user['email'],
+    ))
+
+
+def _add_attr_to_xml(attrs, node) -> None:
+    for attr, fields in attrs.items():
+        attr_node = etree.SubElement(node, attr)
+        _handle_fields(fields, attr_node, attr)
+
+
+def _handle_fields(fields, attr_node, attr) -> None:
+    for field in fields:  # type: ignore
+        field_node = etree.SubElement(attr_node, attr[:-1])
+        for field_name, field_value in field.items():
+            etree.SubElement(
+                field_node,
+                field_name,
+            ).text = str(field_value)

homeworks/relby/6/emails.csv

@@ -0,0 +1,2 @@
+"Sincere@april.biz","Shanna@melissa.tv","anastasia.net"
+"kudinov.nikita@gmail.com","Lucio_Hettinger@annie.ca"

homeworks/relby/6/mypy.ini

@@ -0,0 +1,2 @@
+[mypy]
+ignore_missing_imports = True