adii1823
71 lines
1.7 KiB
Python
71 lines
1.7 KiB
Python
# api.py
|
|
import os
|
|
import csv
|
|
from copy import deepcopy
|
|
|
|
from marshmallow import Schema, fields, pre_load
|
|
from marshmallow.validate import Length, Range
|
|
|
|
|
|
class UserSchema(Schema):
|
|
"""Represent a *valid* user. """
|
|
|
|
email = fields.Email(required=True)
|
|
name = fields.Str(required=True, validate=Length(min=1))
|
|
age = fields.Int(
|
|
required=True, validate=Range(min=18, max=65)
|
|
)
|
|
role = fields.Str()
|
|
|
|
@pre_load()
|
|
def strip_name(self, data, **kwargs):
|
|
data_copy = deepcopy(data)
|
|
|
|
try:
|
|
data_copy['name'] = data_copy['name'].strip()
|
|
except (AttributeError, KeyError, TypeError):
|
|
pass
|
|
|
|
return data_copy
|
|
|
|
|
|
schema = UserSchema()
|
|
|
|
|
|
def export(filename, users, overwrite=True):
|
|
"""Export a CSV file.
|
|
|
|
Create a CSV file and fill with valid users. If `overwrite`
|
|
is False and file already exists, raise IOError.
|
|
"""
|
|
if not overwrite and os.path.isfile(filename):
|
|
raise IOError(f"'{filename}' already exists.")
|
|
|
|
valid_users = get_valid_users(users)
|
|
write_csv(filename, valid_users)
|
|
|
|
|
|
def get_valid_users(users):
|
|
"""Yield one valid user at a time from users. """
|
|
yield from filter(is_valid, users)
|
|
|
|
|
|
def is_valid(user):
|
|
"""Return whether or not the user is valid. """
|
|
return not schema.validate(user)
|
|
|
|
|
|
def write_csv(filename, users):
|
|
"""Write a CSV given a filename and a list of users.
|
|
|
|
The users are assumed to be valid for the given CSV structure.
|
|
"""
|
|
fieldnames = ['email', 'name', 'age', 'role']
|
|
|
|
with open(filename, 'w', newline='') as csvfile:
|
|
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
|
|
for user in users:
|
|
writer.writerow(user)
|