<jupyter_start><jupyter_text>MetaprogrammingDynamic attributes and properties<jupyter_code># osconfeed.py
from urllib.request import urlopen
import warnings
import os
import json
import ssl

URL = 'http://www.oreilly.com/pub/sc/osconfeed'
JSON = 'data/osconfeed.json'
gcontext = ssl.SSLContext(ssl.PROTOCOL_TLSv1)

def load():
    if not os.path.exists(JSON):
        msg = 'downloading {} to {}'.format(URL, JSON)
        warnings.warn(msg)
        with urlopen(URL, context=gcontext) as remote, open(JSON, 'wb') as local:
            local.write(remote.read())
            
        with open(JSON) as fp:
            return json.load(fp)

feed = load()
sorted(feed['Schedule'].keys())
import json
with open('data/osconfeed.json') as fp:
    feed = json.load(fp)
    
sorted(feed['Schedule'].keys())
for key, value in sorted(feed['Schedule'].items()):
    print('{:3} {}'.format(len(value), key))
feed['Schedule']['speakers'][356]['name']
feed['Schedule']['speakers'][-1]['name']
feed['Schedule']['speakers'][356]['serial']
feed['Schedule']['events'][40]['name']
feed['Schedule']['events'][40]['speakers']<jupyter_output><empty_output><jupyter_text>Exploring JSON-like data with dynamic attributes<jupyter_code>import json

def loadf():
    with open('data/osconfeed.json') as fp:
        feed = json.load(fp)
    return feed
    
raw_feed = loadf()
# explore0.py: turn a JSON dataset into a FrozenJSON
from collections import abc

class FrozenJSON:
    """
    A read-only facade for navigating a JSON-like object
    using attribute notation
    """
    
    def __init__(self, mapping):
        self.__data = dict(mapping)
        
    def __getattr__(self, name):
        if hasattr(self.__data, name):
            return getattr(self.__data, name)
        else:
            return FrozenJSON.build(self.__data[name])
        
    @classmethod
    def build(cls, obj):
        if isinstance(obj, abc.Mapping):
            return cls(obj)
        elif isinstance(obj, abc.MutableSequence):
            return [cls.build(item) for item in obj]
        else:
            return obj
feed = FrozenJSON(raw_feed)
len(feed.Schedule.speakers)
sorted(feed.Schedule.keys())
for key, value in sorted(feed.Schedule.items()):
    print('{:3} {}'.format(len(value), key))
feed.Schedule.speakers[-1].name
talk = feed.Schedule.events[40]
talk
type(talk)
talk.name
talk.speakers
talk.flavor
# explore1.py
from collections import abc

class FrozenJSON:
    """
    A read-only facade for navigating a JSON-like object
    using attribute notation
    """
    
    def __init__(self, mapping):
        self.__data = {}
        for key, value in mapping.items():
            if keyword.iskeyword(key):
                key += '_'
            self.__data[key] = value
        
    def __getattr__(self, name):
        if hasattr(self.__data, name):
            return getattr(self.__data, name)
        else:
            return FrozenJSON.build(self.__data[name])
        
    @classmethod
    def build(cls, obj):
        if isinstance(obj, abc.Mapping):
            return  cls(obj)
        elif isinstance(obj, abc.MutableSequence):
            return [cls.build(item) for item in obj]
        else:
            return obj
# explore2.py: using __new__ instead of build to construct new objects
from collections import abc

class FrozenJSON:
    
    def __new__(cls, arg):
        if isinstance(arg, abc.Mapping):
            return super().__new__(cls)
        elif isinstance(arg, abc.MutableSequence):
            return [cls(item) for item in arg]
        else:
            return arg
        
    def __init__(self, mapping):
        self.__data = {}
        for key, value in mapping.items():
            if iskeyword(key):
                key += '_'
            self.__data[key] = value
            
    def __getattr__(self, name):
        if hasattr(self.__data, name):
            return getattr(self.__data, name)
        else:
            return FrozenJSON(self.__data[name])
## Schedule1.py 
# feed with shelve
import shelve
import warnings
import json

DB_NAME = 'data/schedule1_db'
CONFERENCE = 'conference.115'

class Record:
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)

def load():
    with open('data/osconfeed.json') as fp:
        return json.load(fp)
        
def load_db(db):
    raw_data = load()
    warnings.warn('loading' + DB_NAME)
    for collection, rec_list in raw_data['Schedule'].items():
        record_type = collection[:-1]
        for record in rec_list:
            key = '{}.{}'.format(record_type, record['serial'])
            record['serial'] = key
            db[key] = Record(**record)
            
db = shelve.open(DB_NAME)
if CONFERENCE not in db:
    load_db(db)

speaker = db['speaker.3471']
speaker.name, speaker.twitter
## Schedule2.py
import warnings
import inspect
import json

DB_NAME = 'data/schedule2.db'

class Record:
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)
        
    def __eq__(self, other):
        if isinstance(other, Record):
            return self.__dict__ == other.__dict__
        else:
            return NotImplemented

class MissingDatabaseError(RuntimeError):
    """Raised when a database is required but was not set."""
    
class DbRecord(Record):
    
    __db = None
    
    @staticmethod
    def set_db(db):
        DbRecord.__db = db
        
    @staticmethod
    def get_db():
        return DbRecord.__db
    
    @classmethod
    def fetch(cls, ident):
        db = cls.get_db()
        try:
            return db[ident]
        except TypeError:
            if db is None:
                msg = "database not set; call '{}.set_db(my_db)'"
                raise MissingDatabaseError(msg.format(cls.__name__))
            else:
                raise
                
    def __repr__(self):
        if hasattr(self, 'serial'):
            cls_name = self.__class__.__name__
            return '<{} serial={!r}>'.format(cls_name, self.serial)
        else:
            return super().__repr__()
        
class Event(DbRecord):
    
    @property
    def venue(self):
        key = 'venue.{}'.format(self.venue_serial)
        return self.__class__.fetch(key)
    
    @property
    def speakers(self):
        if not hasattr(self, '_speaker_objs'):
            spkr_serials = self.__dict__['speakers']
            fetch = self.__class__.fetch
            self._speaker_objs = [fetch('speaker.{}'.format(key)) for key in spkr_serials]
        return self._speaker_objs
    
    def __repr__(self):
        if hasattr(self, 'name'):
            cls_name = self.__class__.__name__
            return '<{} {!r}>'.format(cls_name, self.name)
        else:
            return super().__repr__()
        
def load():
    with open('data/osconfeed.json') as fp:
        return json.load(fp)

def load_db(db):
    raw_data = load()
    warnings.warn('loading' + DB_NAME)
    for collection, rec_list in raw_data['Schedule'].items():
        record_type = collection[:-1]
        cls_name = record_type.capitalize()
        cls = globals().get(cls_name, DbRecord)
        if inspect.isclass(cls) and issubclass(cls, DbRecord):
            factory = cls
        else:
            factory = DbRecord
        for record in rec_list:
            key = '{}.{}'.format(record_type, record['serial'])
            record['serial'] = key
            db[key] = Record(**record)
import shelve
db = shelve.open(DB_NAME)
load_db(db)
DbRecord.set_db(db)
event = DbRecord.fetch('event.33950')
## Using a property for attribute validation
## LineItem take#1
class LineItem:
    
    def __init__(self, description, weight, price):
        self.description = description
        self.weight = weight
        self.price = price
        
    def subtotal(self):
        return self.weight * self.price
    
raisins = LineItem('Golden raisins', 10, 6.95)
raisins.subtotal()
raisins.weight = -20 #garbage in
raisins.subtotal()  #garbage out
## LineItem take#2
class LineItem:
    
    def __init__(self, description, weight, price):
        self.description = description
        self.weight = weight
        self.price = price
    
    def subtotal(self):
        return self.weight * self.price
    
    @property
    def weight(self):
        return self.__weight
    
    @weight.setter
    def weight(self, value):
        if value > 0:
            self.__weight = value
        else:
            raise ValueError('value must be > 0')
            
walnuts = LineItem('walnuts', 2, 10.00)
walnuts.subtotal()
walnuts.weight
walnuts.weight = 3
walnuts.weight
walnuts.subtotal()
## LineItem a proper look
class LineItem:
    
    def __init__(self, description, weight, price):
        self.description = description
        self.weight = weight
        self.price = price
        
    def subtotal(self):
        return self.weight * self.price
    
    def get_weight(self):
        return self.__weight
    
    def set_weight(self, value):
        if value > 0:
            self.__weight = value
        else:
            raise ValueError('value must be > 0')

    weight = property(get_weight, set_weight)
## Properties override instance attributes
class Class:
    data = 'the class data attr'
    @property
    def prop(self):
        return 'the prop value'
    
obj = Class()
vars(obj) 
obj.__dict__
obj.data
obj.data ='bar'
obj.data
vars(obj)
Class.data
Class.prop
obj.prop
obj.prop = 'foo'
obj.__dict__['prop'] = 'foo'
vars(obj)
obj.prop
Class.prop = 'baz'
obj.prop
Class.data = property(lambda self: 'the "data" prop value')
obj.data
del Class.data
obj.data
vars(obj)
## A property factory
def quantity(storage_name):
    
    def qty_getter(instance):
        return instance.__dict__[storage_name]
    
    def qty_setter(instance, value):
        if value > 0:
            instance.__dict__[storage_name] = value
        else:
            raise ValueError('value must be > 0')
            
    return property(qty_getter, qty_setter)

class LineItem:
    weight = quantity('weight')
    price = quantity('price')
    
    def __init__(self, description, weight, price):
        self.description = description
        self.weight = weight
        self.price = price
        
    def subtotal(self):
        return self.weight * self.price

nutmeg = LineItem('Moluccan nutmeg', 8, 13.95)
nutmeg.weight, nutmeg.price
## Handling attribute deletion
class BlackKnight:
    def __init__(self):
        self.members = ['an arm', 'another arm', 'a leg', 'another leg']
        self.phrases = ["'Til but a scratch.'", "It's just a flesh wound.",
                       "I'm invincible!", "All right, we'll call it a draw"]
        
    @property
    def member(self):
        print('next member is:')
        return self.members[0]
    
    @member.deleter
    def member(self):
        text = 'BLACK KNIGHT (loses {}\n -- {})'
        print(text.format(self.members.pop(0), self.phrases.pop(0)))
knight = BlackKnight()
knight.member
del knight.member
del knight.member
del knight.member
del knight.member<jupyter_output>BLACK KNIGHT (loses another leg
 -- All right, we'll call it a draw)
{'fraction_non_alphanumeric': 0.11721215399873772, 'fraction_numerical': 0.006040934090704175, 'mean_word_length': 2.4055879643844027, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 16, 'https://': 0, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '317051', 'license': 'Apache-2.0', 'n_tokens_mistral': 3420, 'n_tokens_neox': 3361, 'n_words': 867, 'path': 'notebook/fluent_ch19.ipynb', 'repo_name': 'Lin0818/py-study-notebook', 'size': '11091'}