Skip to content

Instantly share code, notes, and snippets.

View brianrodri's full-sized avatar
😎

Brian Rodriguez brianrodri

😎
View GitHub Profile
<!DOCTYPE html>
<html>
<head>
<title>Page Title</title>
</head>
<body>
</body>
</html>
<!DOCTYPE html>
<html>
<head>
<title>Page Title</title>
</head>
<body>
</body>
</html>
<!DOCTYPE html>
<html>
<head>
<title>Do Now 4.1.0</title>
</head>
<body>
<div id="green"></div>
<div id="yellow"></div>
<div id="blue"></div>
<div id="red"></div>
<!DOCTYPE html>
<html>
<head>
<title>Do Now 4.1.0</title>
</head>
<body>
<div id="green"></div>
<div id="yellow"></div>
<div id="blue"></div>
<div id="red"></div>
<!DOCTYPE html>
<html>
<head>
<title>Mini-Activity 4.1.0</title>
</head>
<body>
<div id="green"></div>
<div id="yellow"></div>
<div id="blue"></div>
<div id="red"></div>
class BaseHumanMaintainedModel(BaseModel):
"""A model that tracks the last time it was updated by a human.
When the model is updated by a human, use `self.put_by_human()` to store it.
For example: after the title of an exploration is changed by its creator.
Otherwise, when the model is updated by some non-human or automated process,
use `self.put()`. For example: by a one-off job updating the schema version
of a property.
LATEST_VERSION_SENTINAL = object()
MIGRATION_FUNCTIONS_REGISTRY = {
exp_models.ExplorationModel: {
1: exp_services.migrate_schema_v1_to_v2,
2: exp_services.migrate_schema_v2_to_v3,
3: LATEST_VERSION_SENTINAL,
},
}
class SchemaMigrationTransform(beam.PTransform):
class ModelValidationError(object):
@property
def key(self):
return self.__class__.__name__
@property
def message(self):
return None

Oppia and Apache Beam

Apache Beam is used by Oppia to perform large-scale operations on the datastore. There are two types of operations:

  • Batch: Operations that are designed to be executed once on the "current state" of the datastore.

    Examples:

    • Count the # of models in the datastore
  • Update the StringProperty values of all models with a specific kind

# FILE: jobs/blog_validation_jobs.py
from jobs import base_jobs
from jobs.io import ndb_io
from jobs.types import blog_validation_errors
import apache_beam as beam
class BlogUniquenessJob(base_jobs.JobBase):