ch10

ch10/api.py

@@ -0,0 +1,70 @@
+# 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)