GleeTube

Type-safe Gleam client for the YouTube Data API v3. Invalid API calls fail at compile time, not runtime.

Features

• Type-safe parts and filters -- each resource has its own Part and Filter types, enforced by the compiler
• Result-based error handling -- all API calls return Result(Response, GleeTubeError), no panics
• Pipe-first API -- client |> videos.list(parts, filter) reads naturally
• Automatic pagination -- list_all fetches every page with tail-recursive accumulation
• Pluggable HTTP transport -- built-in httpc adapter, optional hackney adapter with proxy support
• Full API coverage -- 20 YouTube API resources with list, insert, update, delete, and special operations

Installation

gleam add gleetube@1

Quick Start

import gleam/io
import gleam/option.{None}
import gleetube
import gleetube/resource/channels

pub fn main() {
  let client = gleetube.new("YOUR_API_KEY")

  let assert Ok(resp) =
    client
    |> channels.list(
      parts: [channels.Snippet, channels.Statistics],
      filter: channels.ById(["UC_x5XG1OV2P6uZZ5FSM9Ttw"]),
      max_results: None,
      page_token: None,
      hl: None,
    )

  io.debug(resp.items)
}

Type Safety

Parts are typed per resource

Each resource defines its own Part type. Passing a ChannelPart to a video function is a compile error:

import gleetube/resource/videos

// Compiles -- Snippet and Statistics are valid VideoPart variants
videos.list(client,
  parts: [videos.Snippet, videos.Statistics],
  filter: videos.ById(["dQw4w9WgXcQ"]),
  // ...
)

Filters are union types

A filter is required for list operations. The type system ensures exactly one valid filter is provided:

import gleetube/resource/channels

// By ID
channels.list(client,
  parts: [channels.Snippet],
  filter: channels.ById(["UC_x5XG1OV2P6uZZ5FSM9Ttw"]),
  // ...
)

// By YouTube handle
channels.list(client,
  parts: [channels.Snippet],
  filter: channels.ByHandle("@GoogleDevelopers"),
  // ...
)

// Authenticated user's own channel (requires OAuth2)
channels.list(client,
  parts: [channels.Snippet],
  filter: channels.Mine,
  // ...
)

Usage

Videos

import gleam/option.{None, Some}
import gleetube/resource/videos

// List by IDs
let assert Ok(resp) =
  client
  |> videos.list(
    parts: [videos.Snippet, videos.Statistics, videos.ContentDetails],
    filter: videos.ById(["dQw4w9WgXcQ"]),
    hl: None, max_height: None, max_results: None, max_width: None,
    on_behalf_of_content_owner: None, page_token: None,
    region_code: None, video_category_id: None,
  )

// Most popular by region
let assert Ok(resp) =
  client
  |> videos.list(
    parts: [videos.Snippet, videos.Statistics],
    filter: videos.ByChart(videos.MostPopular),
    hl: None, max_height: None, max_results: Some(10), max_width: None,
    on_behalf_of_content_owner: None, page_token: None,
    region_code: Some("US"), video_category_id: None,
  )

// Rate a video (requires OAuth2)
let assert Ok(Nil) =
  client |> videos.rate(video_id: "dQw4w9WgXcQ", rating: videos.Like)

Search

import gleam/option.{None, Some}
import gleetube/resource/search

let assert Ok(resp) =
  client
  |> search.list(
    filter: search.NoFilter,
    q: Some("gleam programming"),
    max_results: Some(10),
    order: Some(search.Relevance),
    safe_search: Some(search.Moderate),
    type_: Some("video"),
    // remaining optional params as None ...
    channel_id: None, channel_type: None, event_type: None,
    location: None, location_radius: None,
    on_behalf_of_content_owner: None, page_token: None,
    published_after: None, published_before: None,
    region_code: None, relevance_language: None,
    topic_id: None, video_caption: None, video_category_id: None,
    video_definition: None, video_dimension: None,
    video_duration: None, video_embeddable: None,
    video_license: None, video_paid_product_placement: None,
    video_syndicated: None, video_type: None,
  )

Playlists

import gleam/option.{None, Some}
import gleetube/resource/playlists

let assert Ok(resp) =
  client
  |> playlists.list(
    parts: [playlists.Snippet, playlists.ContentDetails],
    filter: playlists.ByChannelId("UC_x5XG1OV2P6uZZ5FSM9Ttw"),
    hl: None, max_results: Some(25),
    on_behalf_of_content_owner: None,
    on_behalf_of_content_owner_channel: None,
    page_token: None,
  )

Pagination

Every list function supports manual pagination via page_token. Each resource also provides list_all to fetch all pages automatically:

import gleam/option.{None}
import gleetube/resource/channels

let assert Ok(all_channels) =
  client
  |> channels.list_all(
    parts: [channels.Snippet],
    filter: channels.ByHandle("@GoogleDevelopers"),
    hl: None,
  )

Limit the total number of items with pagination.list_up_to:

import gleetube/pagination

let assert Ok(first_100) =
  pagination.list_up_to(fetch_page, max_count: 100)

Convenience API

The gleetube/api module provides high-level wrappers with sensible defaults:

import gleetube/api

let assert Ok(resp) = api.get_channel_info(client, ["UC_x5XG1OV2P6uZZ5FSM9Ttw"])
let assert Ok(resp) = api.get_video_by_id(client, ["dQw4w9WgXcQ"])
let assert Ok(resp) = api.search_by_keywords(client, "gleam lang", 10)
let assert Ok(resp) = api.get_playlist_items(client, "PLRqwX-V7Uu6ZiZxtDDRCi6uhfTH4FilpH")
let assert Ok(resp) = api.get_comment_threads(client, "dQw4w9WgXcQ")
let assert Ok(resp) = api.get_i18n_languages(client)
let assert Ok(resp) = api.get_video_categories(client, "US")

OAuth2

import gleam/option
import gleam/result
import gleetube
import gleetube/oauth2

let oauth_config = oauth2.new(
  client_id: "YOUR_CLIENT_ID",
  client_secret: "YOUR_CLIENT_SECRET",
  redirect_uri: "http://localhost:8080/callback",
)

// Generate authorization URL -- redirect the user here
let auth_url = oauth2.authorize_url(
  oauth_config,
  access_type: option.Some("offline"),
  state: option.None,
  login_hint: option.None,
  prompt: option.Some(oauth2.Consent),
)

// After the user authorizes, exchange the code for a client
use client <- result.try(
  gleetube.new_with_oauth(oauth_config, code: "AUTH_CODE")
)

// Refresh an expired token
use new_token <- result.try(
  oauth2.refresh_token(oauth_config, refresh_token: "REFRESH_TOKEN")
)

// Revoke a token
let assert Ok(Nil) = oauth2.revoke_token(token: "TOKEN")

Configuration

Custom timeout

import gleetube
import gleetube/auth
import gleetube/config

let client =
  auth.api_key("YOUR_KEY")
  |> config.new()
  |> config.with_timeout(10_000)
  |> gleetube.new_with_config()

Proxy (hackney adapter)

import gleetube
import gleetube/adapter/hackney_adapter
import gleetube/auth
import gleetube/config

let opts =
  hackney_adapter.new()
  |> hackney_adapter.with_proxy("http://proxy:8080")
  |> hackney_adapter.with_proxy_auth("user", "pass")

let client =
  auth.api_key("YOUR_KEY")
  |> config.new()
  |> config.with_transport(
    hackney_adapter.transport(opts),
    hackney_adapter.transport_bits(opts),
  )
  |> gleetube.new_with_config()

Error Handling

All API calls return Result(Response, GleeTubeError). Pattern match on the error variants:

import gleam/io
import gleetube/error.{ApiError, AuthError, DecodeError, HttpError}

case videos.list(client, ...) {
  Ok(resp) -> io.debug(resp.items)
  Error(ApiError(status: 403, message: msg, ..)) ->
    io.println("Forbidden: " <> msg)
  Error(HttpError(message: msg)) ->
    io.println("Network error: " <> msg)
  Error(AuthError(message: msg)) ->
    io.println("Auth failed: " <> msg)
  Error(DecodeError(message: msg)) ->
    io.println("Decode error: " <> msg)
  Error(_) -> io.println("Other error")
}

Supported Resources

Development

gleam build          # compile
gleam test           # run all tests
gleam format         # format code
gleam docs build     # generate docs

License

BlueOak-1.0.0