mcp_layer.py•1.82 kB
import ibis
from boring_semantic_layer import MCPSemanticModel, to_semantic_table
# Copy the public data set to your GCP Project
con = ibis.bigquery.connect(
project_id= "bigquery-public-data",
dataset_id= "austin_bikeshare",
location= "US",
)
bikeshare_trips = con.table("bikeshare_trips")
# here we define dimensions and metrics
bikeshare_trips_model =(
to_semantic_table(bikeshare_trips, name="Bikseshare Trips")
.with_dimensions(
trip_id={
"expr": lambda t: t.trip_id,
"dscription":"Numeric ID of bike trip and primary key of the table"
},
subscriber_type={
"expr": lambda t: t.subscriber_type,
"dscription":"Type of the Subscriber"
},
bike_id={
"expr": lambda t: t.bike_id,
"dscription":"ID of bike used"
},
bike_type={
"expr": lambda t: t.bike_type,
"dscription":"Type of bike used"
},
start_station_name={
"expr": lambda t: t.start_station_name,
"dscription":"Station name for start station"
},
end_station_name={
"expr": lambda t: t.end_station_name,
"dscription":"Station name for end station"
},
)
.with_measures(
total_trip_count={
"expr":lambda t: t.trip_id.count(),
"description": "Total amount of trips"
},
trip_duration={
"expr":lambda t: t.duration_minutes.sum(),
"description": "Time of trip in minutes"
}
)
)
server = MCPSemanticModel(
models={
"Bikeshare Trips": bikeshare_trips_model
},
name="Every row resembles a unique trip with a start and end station and the used bike."
)
if __name__ == "__main__":
server.run()