metaprogramming examples

metaprog/oscon-schedule/explore0.py

@@ -0,0 +1,43 @@
+"""
+explore0.py: Script to download and explore the OSCON schedule feed
+
+    >>> feed = load_json()
+    >>> sorted(feed['Schedule'].keys())
+    ['conferences', 'events', 'speakers', 'venues']
+    >>> for key, value in sorted(feed['Schedule'].items()):
+    ...     print('{:3} {}'.format(len(value), key))
+    ...
+      1 conferences
+    484 events
+    357 speakers
+     53 venues
+    >>> feed['Schedule']['speakers'][-1]['name']
+    'Carina C. Zona'
+    >>> carina = feed['Schedule']['speakers'][-1]
+    >>> carina['twitter']
+    'cczona'
+    >>> feed['Schedule']['events'][40]['name']
+    'There *Will* Be Bugs'
+    >>> feed['Schedule']['events'][40]['speakers']
+    [3471, 5199]
+
+"""
+
+from urllib.request import urlopen
+import warnings
+import os
+import json
+
+URL = 'http://www.oreilly.com/pub/sc/osconfeed'
+JSON_NAME = 'osconfeed.json'
+
+
+def load_json():
+    if not os.path.exists(JSON_NAME):
+        msg = 'downloading {} to {}'.format(URL, JSON_NAME)
+        warnings.warn(msg)
+        with urlopen(URL) as remote, open(JSON_NAME, 'wb') as local:
+            local.write(remote.read())
+
+    with open(JSON_NAME) as fp:
+        return json.load(fp)

metaprog/oscon-schedule/explore1.py

@@ -0,0 +1,69 @@
+"""
+explore.py: Script to download and explore the OSCON schedule feed
+
+    >>> raw_feed = load_json()
+    >>> feed = FrozenJSON(raw_feed)
+    >>> sorted(feed.Schedule.keys())
+    ['conferences', 'events', 'speakers', 'venues']
+    >>> for key, value in sorted(feed.Schedule.items()):
+    ...     print('{:3} {}'.format(len(value), key))
+    ...
+      1 conferences
+    484 events
+    357 speakers
+     53 venues
+    >>> feed.Schedule.speakers[-1].name
+    'Carina C. Zona'
+    >>> carina = feed.Schedule.speakers[-1]
+    >>> carina.twitter
+    'cczona'
+    >>> feed.Schedule.events[40].name
+    'There *Will* Be Bugs'
+    >>> feed.Schedule.events[40].speakers
+    [3471, 5199]
+
+"""
+
+from urllib.request import urlopen
+import warnings
+import os
+import json
+from collections import abc
+
+URL = 'http://www.oreilly.com/pub/sc/osconfeed'
+JSON_NAME = 'osconfeed.json'
+
+
+def load_json():
+    if not os.path.exists(JSON_NAME):
+        msg = 'downloading {} to {}'.format(URL, JSON_NAME)
+        warnings.warn(msg)
+        with urlopen(URL) as remote, open(JSON_NAME, 'wb') as local:
+            local.write(remote.read())
+
+    with open(JSON_NAME) as fp:
+        return json.load(fp)
+
+
+class FrozenJSON:
+    """A read-only façade 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

metaprog/oscon-schedule/explore2.py

@@ -0,0 +1,68 @@
+"""
+explore2.py: Script to download and explore the OSCON schedule feed
+
+    >>> raw_feed = load_json()
+    >>> feed = FrozenJSON(raw_feed)
+    >>> sorted(feed.Schedule.keys())
+    ['conferences', 'events', 'speakers', 'venues']
+    >>> for key, value in sorted(feed.Schedule.items()):
+    ...     print('{:3} {}'.format(len(value), key))
+    ...
+      1 conferences
+    484 events
+    357 speakers
+     53 venues
+    >>> feed.Schedule.speakers[-1].name
+    'Carina C. Zona'
+    >>> carina = feed.Schedule.speakers[-1]
+    >>> carina.twitter
+    'cczona'
+    >>> feed.Schedule.events[40].name
+    'There *Will* Be Bugs'
+    >>> feed.Schedule.events[40].speakers
+    [3471, 5199]
+
+"""
+
+from urllib.request import urlopen
+import warnings
+import os
+import json
+from collections import abc
+
+URL = 'http://www.oreilly.com/pub/sc/osconfeed'
+JSON_NAME = 'osconfeed.json'
+
+
+def load_json():
+    if not os.path.exists(JSON_NAME):
+        msg = 'downloading {} to {}'.format(URL, JSON_NAME)
+        warnings.warn(msg)
+        with urlopen(URL) as remote, open(JSON_NAME, 'wb') as local:
+            local.write(remote.read())
+
+    with open(JSON_NAME) as fp:
+        return json.load(fp)
+
+
+class FrozenJSON:
+    """A read-only façade for navigating a JSON-like object
+       using attribute notation
+    """
+
+    def __new__(cls, arg):
+        if isinstance(arg, abc.Mapping):
+            return super().__new__(cls)
+        elif isinstance(arg, abc.MutableSequence):
+            return [FrozenJSON(item) for item in arg]
+        else:
+            return arg
+
+    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(self._data[name])

metaprog/oscon-schedule/osconfeed-sample.json

@@ -0,0 +1,32 @@
+{ "Schedule": 
+  { "conferences": [{"serial": 115 }],
+    "events": [
+      { "serial": 34505,
+        "name": "Why Schools Don't Use Open Source to Teach Programming",
+        "event_type": "40-minute conference session",
+        "time_start": "2014-07-23 11:30:00",
+        "time_stop": "2014-07-23 12:10:00",
+        "venue_serial": 1462,
+        "description": "Aside from the fact that high school programming curricula often require proprietary IDEs, they also don't involve examining any source code from Open Source software projects. What changes would be required in programming curricula to incorporate Open Source? And is that a desirable objective?",
+        "website_url": "http://oscon.com/oscon2014/public/schedule/detail/34505", 
+        "speakers": [157509],
+        "categories": ["Education"] }
+    ],
+    "speakers": [
+      { "serial": 157509,
+        "name": "Robert Lefkowitz",
+        "photo": null,
+        "url": "http://sharewave.com/",
+        "position": "CTO",
+        "affiliation": "Sharewave",
+        "twitter": "sharewaveteam",
+        "bio": "Robert 'r0ml' Lefkowitz is the CTO at Sharewave, a startup building an investor management portal. This year, he was a resident at the winter session of Hacker School. He is a Distinguished Engineer of the ACM." }
+    ],
+    "venues": [
+      { "serial": 1462,
+        "name": "F151",
+        "category": "Conference Venues" }
+    ]
+  }
+}
+    

metaprog/oscon-schedule/schedule.py

@@ -0,0 +1,36 @@
+"""
+
+>>> db = shelve.open(DB_NAME)
+>>> if CONFERENCE not in db: load_db(db)
+>>> event = db['event.33950']
+>>> record = db['speaker.3471']
+>>> record.name
+'Anna Martelli Ravenscroft'
+>>> record.twitter
+'annaraven'
+>>> db.close()
+
+"""
+
+import warnings
+import shelve
+
+from explore import load_json
+
+DB_NAME = 'schedule_db'
+CONFERENCE = 'conference.115'
+
+
+class Record:
+    def __init__(self, mapping):
+        self.__dict__.update(mapping)
+
+
+def load_db(db):
+    raw_data = load_json()
+    warnings.warn('loading ' + DB_NAME)
+    for collection, rec_list in raw_data['Schedule'].items():
+        rec_type = collection[:-1]
+        for fields in rec_list:
+            key = '{}.{}'.format(rec_type, fields['serial'])
+            db[key] = Record(fields)

metaprog/oscon-schedule/schedule2.py

@@ -0,0 +1,62 @@
+import warnings
+import inspect
+
+from explore import load_json
+
+DB_NAME = 'schedule2_db'
+CONFERENCE = 'conference.115'
+
+
+class Record:
+    def __init__(self, mapping):
+        self.__dict__.update(mapping)
+
+    def __eq__(self, other):
+        if isinstance(other, Record):
+            return self.__dict__ == other.__dict__
+        else:
+            return NotImplemented
+
+    def __repr__(self):
+        if hasattr(self, 'name'):
+            ident = repr(self.name)
+        else:
+            ident = 'object at ' + hex(id(self))
+        cls_name = self.__class__.__name__
+        return '<{} {}>'.format(cls_name, ident)
+
+
+class Event(Record):
+
+    @classmethod
+    def set_db(cls, db):
+        cls._db = db
+
+    @property
+    def venue(self):
+        key = self.venue_serial
+        return self._db['venue.{}'.format(key)]
+
+    @property
+    def speakers(self):
+        spkr_serials = self.__dict__['speakers']
+        if not hasattr(self, '_speaker_refs'):
+            self._speaker_refs = [self._db['speaker.{}'.format(key)]
+                                  for key in spkr_serials]
+        return self._speaker_refs
+
+
+def load_db(db):
+    raw_data = load_json()
+    warnings.warn('loading ' + DB_NAME)
+    for collection, rec_list in raw_data['Schedule'].items():
+        rec_type = collection[:-1]
+        for fields in rec_list:
+            cls_name = rec_type.capitalize()
+            cls = globals().get(cls_name, Record)
+            if inspect.isclass(cls) and issubclass(cls, Record):
+                record = cls(fields)
+            else:
+                Record(fields)
+            key = '{}.{}'.format(rec_type, fields['serial'])
+            db[key] = record

metaprog/oscon-schedule/test_schedule.py

@@ -0,0 +1,37 @@
+import shelve
+import pytest
+
+import schedule
+
+
+@pytest.yield_fixture
+def db():
+    with shelve.open(schedule.DB_NAME) as the_db:
+        if schedule.CONFERENCE not in the_db:
+            schedule.load_db(the_db)
+        yield the_db
+
+
+def test_record_class():
+    rec = schedule.Record({'spam': 99, 'eggs': 12})
+    assert rec.spam == 99
+    assert rec.eggs == 12
+
+
+def test_conference_record(db):
+    assert schedule.CONFERENCE in db
+
+
+def test_speaker_record(db):
+    speaker = db['speaker.3471']
+    assert speaker.name == 'Anna Martelli Ravenscroft'
+
+
+def test_event_record(db):
+    event = db['event.33950']
+    assert event.name == 'There *Will* Be Bugs'
+
+
+def test_event_venue(db):
+    event = db['event.33950']
+    assert event.venue_serial == 1449

metaprog/oscon-schedule/test_schedule2.py

@@ -0,0 +1,61 @@
+import shelve
+import pytest
+
+import schedule2 as schedule
+
+
+@pytest.yield_fixture
+def db():
+    with shelve.open(schedule.DB_NAME) as the_db:
+        if schedule.CONFERENCE not in the_db:
+            schedule.load_db(the_db)
+        yield the_db
+
+
+def test_record_attr_access():
+    rec = schedule.Record({'spam': 99, 'eggs': 12})
+    assert rec.spam == 99
+    assert rec.eggs == 12
+
+
+def test_record_repr():
+    rec = schedule.Record({'spam': 99, 'eggs': 12})
+    assert repr(rec).startswith('<Record object at 0x')
+    rec2 = schedule.Record({'name': 'Fido'})
+    assert repr(rec2) == "<Record 'Fido'>"
+
+
+def test_conference_record(db):
+    assert schedule.CONFERENCE in db
+
+
+def test_speaker_record(db):
+    speaker = db['speaker.3471']
+    assert speaker.name == 'Anna Martelli Ravenscroft'
+
+
+def test_event_record(db):
+    event = db['event.33950']
+    assert event.name == 'There *Will* Be Bugs'
+
+
+def test_event_venue(db):
+    schedule.Event.set_db(db)
+    event = db['event.33950']
+    assert event.venue_serial == 1449
+    assert event.venue == db['venue.1449']
+    assert event.venue.name == 'Portland 251'
+
+
+def test_event_speakers(db):
+    schedule.Event.set_db(db)
+    event = db['event.33950']
+    assert len(event.speakers) == 2
+    anna_and_alex = [db['speaker.3471'], db['speaker.5199']]
+    assert event.speakers == anna_and_alex
+
+def test_event_no_speakers(db):
+    schedule.Event.set_db(db)
+    event = db['event.36848']
+    assert len(event.speakers) == 0
+