{
  "id": "1b1c4a8c-89aa-4385-8f3e-16b757ac5384",
  "slug": "ref/web/blog-bizspring-co-kr/fastapi와-bigquery-공개데이터로-rest-api-만들기-bizspring-blog",
  "doc_type": "ref",
  "title": "FastAPI와 BigQuery 공개데이터로 REST API 만들기 - BizSpring BLOG",
  "title_en": null,
  "one_liner": "FastAPI와 BigQuery 공개데이터로 REST API 만들기 - BizSpring BLOG 콘텐츠로 건너뛰기 내비게이션 메뉴 인사이트 테크 활용 방법/사례 활용 방법/사례 성공사례 성공 사례 일반 광고/마케팅 에이전시 이커머스 미디어/콘텐츠 금융/핀테크 의",
  "summary_bullets": [
    "FastAPI와 Google BigQuery 공개데이터셋(google_analytics_sample)을 연동해 REST API로 제공하는 방법을 단계별로 설명한다.",
    "Python 가상환경 구성부터 GCP 프로젝트 생성, BigQuery API 활성화, 서비스 계정 인증키 발급까지 개발 환경 준비 과정을 다룬다.",
    "routers→services→repositories→core로 이어지는 계층형 FastAPI 프로젝트 구조와 lifespan을 통한 BigQuery 클라이언트 주입 방식을 소개한다.",
    "Pydantic 모델로 응답 스키마를 정의하고 SQL 인젝션 방지, 파라미터 바인딩을 적용한 Repository/Service/Router 구현 예제를 제공한다.",
    "/ga_sessions/{yyyymmdd} 엔드포인트로 실제 쿼리 결과를 확인하며, 동일 구조를 다른 BigQuery 프로젝트에도 확장 적용할 수 있음을 안내한다."
  ],
  "body_md": "FastAPI와 BigQuery 공개데이터로 REST API 만들기 - BizSpring BLOG\n\n콘텐츠로 건너뛰기\n\n내비게이션 메뉴\n\n인사이트\n\n테크\n\n활용 방법/사례\n\n활용 방법/사례\n\n성공사례\n\n성공 사례 일반\n\n광고/마케팅 에이전시\n\n이커머스\n\n미디어/콘텐츠\n\n금융/핀테크\n\n의료/헬스케어\n\n통신/인터넷\n\n뉴스/트렌드\n\n릴리즈 노트\n\n인터넷트렌드 ↗\n\n웹사이트 ↗\n\n# FastAPI와 BigQuery 공개데이터로 REST API 만들기\n\n2025년 11월 10일 2025년 11월 10일\n테크\n\nGoogle Cloud의 BigQuery는 대규모 데이터를 SQL 기반으로 분석할 수 있는 서비스이며, FastAPI는 Python 기반의 고성능 웹 프레임워크로 RESTful API 구현에 적합합니다. 이번 포스트에서는 Google이 제공하는 공개 데이터셋인 bigquery-public-data.google_analytics_sample을 FastAPI를 통해 외부에서 호출 가능한 API 형태로 변환함으로써, BigQuery 데이터를 서비스 백엔드 수준에서 다루는 과정을 단계별로 살펴보겠습니다.\n\n## 1. 개발 환경 구축 및 GCP 설정\n\nFastAPI와 BigQuery 클라이언트를 실행하기 위한 로컬 개발 환경과 GCP 프로젝트를 설정합니다. Python 가상환경을 만들고 필요한 패키지를 설치한 뒤, BigQuery API가 활성화된 GCP 프로젝트를 준비합니다.\n\n1.1 python 및 라이브러리 설치 https://www.python.org/downloads/ 설치 후 아래 명령어로 버전을 확인합니다.\n\npython –version\n\n1.2 가상환경 생성 및 활성화\n\npython -m venv .venv\n.venv/scripts/activate\n\n1.3 pip 업그레이드 및 라이브러리 설치\n\npython.exe -m pip install --upgrade pip\npip install fastapi[standard] google-cloud-bigquery google-cloud-bigquery-storage pandas python-dotenv pydantic-settings pyarrow\n\n1.4 GCP 프로젝트 생성 https://console.cloud.google.com/ 새 프로젝트를 생성합니다.\n\n1.5 BigQuery API 활성화 API 및 서비스 → 라이브러리 → BigQuery API → 사용 설정\n\n1.6 서비스 계정 생성 IAM 및 관리자 → 서비스 계정 → 서비스 계정 만들기 권한부여: BigQuery 작업 사용자, 데이터 뷰어\n\n1.7 인증 키 발급 생성된 서비스 계정을 클릭 → 키 → 키 추가 → 새 키 만들기 JSON 형식의 비공개 키 파일을 다운로드 후 프로젝트 루트 디렉터리에 저장합니다.\n\n## 2. 데이터 셋 소개\n\nconsole.cloud.google.com/marketplace/product/obfuscated-ga360-data/obfuscated-ga360-data\n\nRest API에 활용할 샘플 데이터는 bigquery-public-data.google_analytics_sample 입니다. Google Merchandise Store의 실제 웹로그 데이터를 기반으로 만들어진 샘플이며, BigQuery에서 누구나 접근할 수 있습니다.\n\nbigquery-public-data.google_analytics_sample.INFORMATION_SCHEMA.COLUMNS\n\n핵심 테이블은 ga_sessions_* 입니다. 날짜 단위로 쪼개져 있으며, 각 세션의 정보가 담겨 있습니다. 스키마는 다소 복잡하지만, 크게 보면 다음과 같은 그룹으로 구성됩니다.\n\n트랜잭션 정보 : 구매 여부, 구매 건수, 매출 금액\n\n디바이스 정보 : 브라우저, 운영체제, 모바일 여부 등\n\n지역 정보 : 국가, 도시, 대륙\n\n행동 정보 : 페이지뷰, 세션 길이 등\n\n## 3. 프로젝트 파일 구조\n\n아래는 FastAPI 애플리케이션의 계층형 구조 예시입니다. 각 디렉터리는 역할별로 모듈화되어 있으며, 의존 방향은 routers → services → repositories → core 순으로 구성됩니다. 현재 예제에서는 app.state.bq를 통해 BigQuery 클라이언트를 전역으로 주입하지만, 규모가 커질 경우 의존성 관리와 리소스 생명주기 제어를 위해 Provider(LOC 컨테이너) 역할을 하는 별도의 모듈이 필요합니다.\n\npytest/\n├── app/\n│ ├── main.py # FastAPI 앱 진입점\n│ │\n│ ├── core/ # BigQuery 클라이언트\n│ │ ├── __init__.py\n│ │ └── bigquery.py\n│ │\n│ ├── models/ # 데이터 모델\n│ │ ├── __init__.py\n│ │ └── ga_sessions.py\n│ │\n│ ├── repositories/ # 데이터 접근 계층\n│ │ ├── __init__.py\n│ │ └── ga_sessions.py\n│ │\n│ ├── services/ # 비즈니스 로직 계층\n│ │ ├── __init__.py\n│ │ └── ga_sessions.py\n│ │\n│ └── routers/ # API 엔드포인트 계층\n│ ├── __init__.py\n│ └── ga_sessions.py\n│\n└── config/ # 설정 파일 (BigQuery 인증 키 파일 위치)\n\n## 4. .env 설정\n\n환경 변수 파일 .env를 통해 BigQuery 자격 증명 경로, 프로젝트 ID, 리전 등의 설정값을 관리합니다. 이 값들은 코드 내에서 os.getenv()로 불러와 클라이언트 초기화에 사용됩니다.\n\nGOOGLE_APPLICATION_CREDENTIALS=\nBIGQUERY_PROJECT_ID=\nBIGQUERY_LOCATION=US\n\n## 5. BigQuery 클라이언트 설정\n\n앱 시작 시 FastAPI의 lifespan 컨텍스트 내에서 BigQuery 클라이언트를 초기화하고, 앱 종료 시 세션을 닫습니다. 이 클라이언트는 Depends를 통해 각 요청 핸들러에서 주입됩니다.\n\n#app/core/bigquery.py\nfrom dotenv import load_dotenv\nfrom contextlib import asynccontextmanager\nfrom google.cloud import bigquery\nimport os\nfrom fastapi import Request\n\nload_dotenv()\n\n@asynccontextmanager\nasync def lifespan(app):\nclient = bigquery.Client.from_service_account_json(\nos.getenv(\"GOOGLE_APPLICATION_CREDENTIALS\"),\nproject=os.getenv(\"BIGQUERY_PROJECT_ID\"),\nlocation=os.getenv(\"BIGQUERY_LOCATION\"),\n)\napp.state.bq = client\nyield\nclient.close()\n\ndef get_bq_client(request: Request):\nreturn request.app.state.bq\n\n## 6. 모델 정의\n\nPydantic 모델을 사용해 BigQuery 쿼리 결과를 구조화합니다. 필요한 주요 필드만 정의하여 응답 스키마를 간결하게 유지하고, 추가 필드는 무시합니다.\n\n# app/models/ga_sessions.py\nfrom typing import Optional, List, Dict, Any\nfrom pydantic import BaseModel, Field, ConfigDict\n\nclass TotalsSlim(BaseModel):\nvisits: Optional[int] = None\nhits: Optional[int] = None\npageviews: Optional[int] = None\ntimeOnSite: Optional[int] = None\n\nclass TrafficSourceSlim(BaseModel):\nsource: Optional[str] = None\nmedium: Optional[str] = None\nreferralPath: Optional[str] = None\n\nclass DeviceSlim(BaseModel):\nbrowser: Optional[str] = None\noperatingSystem: Optional[str] = None\nisMobile: Optional[bool] = None\n\nclass GeoSlim(BaseModel):\ncontinent: Optional[str] = None\ncountry: Optional[str] = None\n\nclass CustomDimension(BaseModel):\nindex: Optional[int] = None\nvalue: Optional[str] = None\n\nclass GaSessionSlim(BaseModel):\nvisitorId: Optional[str] = Field(None, description=\"INT64 → string\")\nvisitId: Optional[str] = Field(None, description=\"INT64 → string\")\nfullVisitorId: Optional[str] = None\n\nvisitNumber: Optional[int] = None\nvisitStartTime: Optional[int] = None\ndate: Optional[str] = None\nchannelGrouping: Optional[str] = None\n\ntotals: Optional[TotalsSlim] = None\ntrafficSource: Optional[TrafficSourceSlim] = None\ndevice: Optional[DeviceSlim] = None\ngeoNetwork: Optional[GeoSlim] = None\ncustomDimensions: Optional[List[CustomDimension]] = None\n\n# 선언하지 않은 필드는 무시\nmodel_config = ConfigDict(extra=\"ignore\")\n\n## 7. Repository 구현: 공개데이터 조회\n\nRepository 계층은 실제 BigQuery SQL을 실행해 데이터를 가져옵니다. 날짜 포맷 검증을 통해 SQL 인젝션이나 잘못된 요청을 방지하고, 파라미터 바인딩으로 limit을 제어합니다.\n\n# app/repositories/ga_sessions.py\nimport re\nfrom typing import Iterable, Dict, Any, List\nfrom google.cloud import bigquery\n\n_YMD_RE = re.compile(r\"^\\d{8}$\")\n\ndef _validate_yyyymmdd(yyyymmdd: str) -> None:\nif not _YMD_RE.fullmatch(yyyymmdd):\nraise ValueError(\"yyyymmdd must be 8 digits like 20160801\")\n\ndef query_sessions(bq: bigquery.Client, yyyymmdd: str, limit: int) -> Iterable[Dict[str, Any]]:\n_validate_yyyymmdd(yyyymmdd)\ntable = f\"`bigquery-public-data.google_analytics_sample.ga_sessions_{yyyymmdd}`\"\n\nsql = f\"\"\"\nSELECT\nCAST(visitorId AS STRING) AS visitorId,\nCAST(visitId AS STRING) AS visitId,\nfullVisitorId,\nvisitNumber,\nvisitStartTime,\ndate,\nchannelGrouping,\ntotals.visits AS totals_visits,\ntotals.hits AS totals_hits,\ntotals.pageviews AS totals_pageviews,\ntotals.timeOnSite AS totals_timeOnSite,\ntrafficSource.source AS trafficSource_source,\ntrafficSource.medium AS trafficSource_medium,\ntrafficSource.referralPath AS trafficSource_referralPath,\ndevice.browser AS device_browser,\ndevice.operatingSystem AS device_operatingSystem,\ndevice.isMobile AS device_isMobile,\ngeoNetwork.continent AS geoNetwork_continent,\ngeoNetwork.country AS geoNetwork_country,\ncustomDimensions AS customDimensions\nFROM {table}\nLIMIT @limit\n\"\"\"\n\njob_config = bigquery.QueryJobConfig(\nquery_parameters=[bigquery.ScalarQueryParameter(\"limit\", \"INT64\", limit)]\n)\nrows = bq.query(sql, job_config=job_config).result()\nfor r in rows:\nyield {k: r.get(k) for k in r.keys()}\n\n## 8. Service\n\nService 계층은 Repository에서 가져온 flat한 쿼리 결과를 Pydantic 모델 형태로 변환합니다. 이 과정을 통해 API 응답이 일관된 구조(GaSessionSlim)를 유지합니다.\n\n# app/services/ga_sessions.py\nfrom typing import Dict, Any, List\nfrom google.cloud.bigquery import Client\nfrom app.models.ga_sessions import GaSessionSlim\nfrom app.repositories.ga_sessions import query_sessions\n\ndef _nest(record: Dict[str, Any]) -> Dict[str, Any]:\nreturn {\n\"visitorId\": record.get(\"visitorId\"),\n\"visitId\": record.get(\"visitId\"),\n\"fullVisitorId\": record.get(\"fullVisitorId\"),\n\"visitNumber\": record.get(\"visitNumber\"),\n\"visitStartTime\": record.get(\"visitStartTime\"),\n\"date\": record.get(\"date\"),\n\"channelGrouping\": record.get(\"channelGrouping\"),\n\"totals\": {\n\"visits\": record.get(\"totals_visits\"),\n\"hits\": record.get(\"totals_hits\"),\n\"pageviews\": record.get(\"totals_pageviews\"),\n\"timeOnSite\": record.get(\"totals_timeOnSite\"),\n},\n\"trafficSource\": {\n\"source\": record.get(\"trafficSource_source\"),\n\"medium\": record.get(\"trafficSource_medium\"),\n\"referralPath\": record.get(\"trafficSource_referralPath\"),\n},\n\"device\": {\n\"browser\": record.get(\"device_browser\"),\n\"operatingSystem\": record.get(\"device_operatingSystem\"),\n\"isMobile\": record.get(\"device_isMobile\"),\n},\n\"geoNetwork\": {\n\"continent\": record.get(\"geoNetwork_continent\"),\n\"country\": record.get(\"geoNetwork_country\"),\n},\n\"customDimensions\": record.get(\"customDimensions\"),\n}\n\ndef to_ga_sessions(records: List[Dict[str, Any]]) -> List[GaSessionSlim]:\nreturn [GaSessionSlim(**_nest(r)) for r in records]\n\ndef get_ga_sessions(bq: Client, yyyymmdd: str, limit: int) -> List[GaSessionSlim]:\nrecords = list(query_sessions(bq, yyyymmdd, limit))\nreturn to_ga_sessions(records)\n\n## 9. Router 및 요청 예시\n\n라우터는 /ga_sessions/{yyyymmdd} 형태의 엔드포인트를 제공하며, limit 쿼리 파라미터로 결과 개수를 제한합니다.\n\n# app/routers/ga_sessions.py\nfrom typing import List\nfrom fastapi import APIRouter, Depends, HTTPException, Query\nfrom google.cloud.bigquery import Client\nfrom app.core.bigquery import get_bq_client\nfrom app.models.ga_sessions import GaSessionSlim\nfrom app.services.ga_sessions import get_ga_sessions\n\nrouter = APIRouter(prefix=\"/ga_sessions\", tags=[\"ga_sessions\"])\n\n@router.get(\"/{yyyymmdd}\", response_model=List[GaSessionSlim])\ndef list_ga_sessions(\nyyyymmdd: str,\nlimit: int = Query(100, ge=1, le=1000),\nbq: Client = Depends(get_bq_client),\n):\ntry:\nreturn get_ga_sessions(bq, yyyymmdd, limit)\nexcept ValueError as ve:\nraise HTTPException(status_code=400, detail=str(ve))\nexcept Exception as e:\nraise HTTPException(status_code=500, detail=\"Query failed\")\n\n## 10. 실행 및 테스트\n\n아래 명령어로 로컬 서버를 실행한 후 브라우저에서 /ga_sessions/20160801 엔드포인트를 호출해 응답을 확인합니다.\n\nuvicorn app.main:app --reload --host 127.0.0.1 --port 8000\n\n/ga_sessions/20160801 호출 결과\n\n이번 포스트에서는 다음 과정을 통해 BigQuery 공개 데이터셋을 REST API 형태로 제공하는 전체 흐름을 구현했습니다. 공개 데이터셋뿐 아니라 다른 BigQuery 프로젝트에도 동일한 구조를 적용할 수 있으며, 인증·캐싱·비동기 쿼리 처리 등을 추가해 확장할 수도 있습니다.\n\nRef. FastAPI 구글 애널리틱스 샘플\n\n최신 마케팅/고객 데이터 활용 사례를 받아보실 수 있습니다.\n비즈스프링 뉴스레터 구독하기 →\n\n## Related Posts via Categories\n\n구글애즈가 정확검색·구문검색 키워드까지 AI 모드에 노출하기 시작했다 — 자동화 상품 없이 AI 답변 화면에 들어가는 첫 실험\nAI가 React 앱을 직접 디버깅 – Agent 기반 디버깅\nChatGPT 광고가 픽셀·전환 API·맞춤 오디언스를 갖췄다 — 답변 화면이 측정 가능한 광고 매체가 되는 순간\nAI 시대, 데이터도 설명이 필요합니다\nMCP 클라이언트는 Authorization Server를 어떻게 찾는가: 공식 MCP TypeScript SDK와 VS Code 구현 비교\nRAG 시대의 GEO: AI가 콘텐츠를 읽는 방식과 스키마의 진짜 역할\nAI 에이전트로 웹 데이터 수집 가능 여부 자동 검증하기\nGA4 Intraday 실시간 테이블, 대시보드 원천 데이터로 바로 사용할 수 있을까?\n워크플로우 오케스트레이션 플랫폼 Temporal 알아보기\niOS 14.5부터 GA4까지, 환경 변화가 가져온 업무 폭증을 해결하는 실무 전략은…?\n\n다음에 대해 검색하기...\n\n최신 글 둘러보기\n\n기존 SEO와 무엇이 같고 무엇이 다른가\n\n콘텐츠가 답인 걸 모르는 사람은 없습니다 — 문제는 지속입니다\n\n광고비를 늘리지 않고 매출을 늘린 회사들은 무엇을 했나\n\n코호트로 비교하면 광고 예산 판단이 달라집니다\n\nGEO를 위한 글쓰기를 위한 작은 노하우\n\n(해외동향) 광고주가 광고플랫폼 리포트에서 자체 측정으로 옮겨가고 있다.\n\n구글애즈가 정확검색·구문검색 키워드까지 AI 모드에 노출하기 시작했다 — 자동화 상품 없이 AI 답변 화면에 들어가는 첫 실험\n\nAI가 React 앱을 직접 디버깅 – Agent 기반 디버깅\n\nChatGPT 광고가 픽셀·전환 API·맞춤 오디언스를 갖췄다 — 답변 화면이 측정 가능한 광고 매체가 되는 순간\n\n마케팅 자동화, 무엇부터 자동화해야 할까?\n\n출처: https://blog.bizspring.co.kr/%ed%85%8c%ed%81%ac/fastapi%ec%99%80-bigquery-%ea%b3%b5%ea%b0%9c%eb%8d%b0%ec%9d%b4%ed%84%b0%eb%a1%9c-rest-api-%eb%a7%8c%eb%93%a4%ea%b8%b0/",
  "related_slugs": [],
  "faq": [
    {
      "a": "Google이 제공하는 bigquery-public-data.google_analytics_sample 데이터셋으로, Google Merchandise Store의 실제 웹로그 데이터를 기반으로 만들어졌습니다. BigQuery에서 누구나 접근할 수 있는 공개 데이터입니다.",
      "q": "이 예제에서 사용하는 공개 데이터셋은 무엇인가요?"
    },
    {
      "a": "GCP 프로젝트를 생성한 뒤 API 및 서비스 라이브러리에서 BigQuery API를 활성화해야 합니다. 이후 서비스 계정을 만들어 BigQuery 작업 사용자와 데이터 뷰어 권한을 부여하고, JSON 형식의 인증 키를 발급받아야 합니다.",
      "q": "BigQuery API를 사용하려면 어떤 준비가 필요한가요?"
    },
    {
      "a": "routers, services, repositories, core, models 디렉터리로 역할별 모듈화되어 있으며, 의존 방향은 routers → services → repositories → core 순입니다. 현재 예제는 app.state.bq로 BigQuery 클라이언트를 전역 주입하지만, 규모가 커지면 별도의 Provider 모듈이 필요할 수 있습니다.",
      "q": "FastAPI 프로젝트는 어떤 구조로 구성되나요?"
    },
    {
      "a": "Repository 계층에서 정규식으로 yyyymmdd 형식을 검증하고, limit 값은 쿼리 파라미터 바인딩 방식으로 처리해 SQL 인젝션이나 잘못된 요청을 방지합니다.",
      "q": "잘못된 날짜 형식이나 SQL 인젝션은 어떻게 방지하나요?"
    },
    {
      "a": "uvicorn app.main:app --reload 명령어로 로컬 서버를 실행한 뒤, 브라우저에서 /ga_sessions/20160801과 같은 엔드포인트를 호출해 응답 결과를 확인할 수 있습니다.",
      "q": "구현한 API는 어떻게 실행하고 테스트하나요?"
    }
  ],
  "jsonld": null,
  "keywords": [],
  "source_id": "https://blog.bizspring.co.kr/%ed%85%8c%ed%81%ac/fastapi%ec%99%80-bigquery-%ea%b3%b5%ea%b0%9c%eb%8d%b0%ec%9d%b4%ed%84%b0%eb%a1%9c-rest-api-%eb%a7%8c%eb%93%a4%ea%b8%b0/",
  "video_url": null,
  "duration_sec": null,
  "thumbnail_url": null,
  "image_url": null,
  "caption": null,
  "solution_slug": null,
  "captured_at": null,
  "width": null,
  "height": null,
  "created_at": "2026-09-06T11:11:02.070275+00:00",
  "updated_at": "2026-09-06T11:11:02.070275+00:00",
  "visibility": "public",
  "source_stage": "1",
  "anonymized": true,
  "source_key": "web-blog",
  "origin": "ingest",
  "locked": false,
  "locked_by": null,
  "locked_at": null,
  "gate_state": {
    "g1": {
      "hits": [],
      "pass": true
    },
    "g2": {
      "pass": true,
      "total": 44,
      "issues": [
        "숫자·규격 같은 검증 가능한 구체가 부족합니다",
        "다른 문서와 겹치는 말이 대부분입니다"
      ],
      "applies": true
    },
    "g3": {
      "pass": true,
      "required": false
    },
    "stage": "1",
    "decided_at": "2026-09-12T15:43:23.696Z"
  },
  "raw_id": "67cb5657-de8f-4110-9615-ce7c8d1d478e",
  "search_text": "FastAPI와 BigQuery 공개데이터로 REST API 만들기 - BizSpring BLOG  FastAPI와 BigQuery 공개데이터로 REST API 만들기 - BizSpring BLOG 콘텐츠로 건너뛰기 내비게이션 메뉴 인사이트 테크 활용 방법/사례 활용 방법/사례 성공사례 성공 사례 일반 광고/마케팅 에이전시 이커머스 미디어/콘텐츠 금융/핀테크 의  FastAPI와 BigQuery 공개데이터로 REST API 만들기 - BizSpring BLOG\n\n콘텐츠로 건너뛰기\n\n내비게이션 메뉴\n\n인사이트\n\n테크\n\n활용 방법/사례\n\n활용 방법/사례\n\n성공사례\n\n성공 사례 일반\n\n광고/마케팅 에이전시\n\n이커머스\n\n미디어/콘텐츠\n\n금융/핀테크\n\n의료/헬스케어\n\n통신/인터넷\n\n뉴스/트렌드\n\n릴리즈 노트\n\n인터넷트렌드 ↗\n\n웹사이트 ↗\n\n# FastAPI와 BigQuery 공개데이터로 REST API 만들기\n\n2025년 11월 10일 2025년 11월 10일\n테크\n\nGoogle Cloud의 BigQuery는 대규모 데이터를 SQL 기반으로 분석할 수 있는 서비스이며, FastAPI는 Python 기반의 고성능 웹 프레임워크로 RESTful API 구현에 적합합니다. 이번 포스트에서는 Google이 제공하는 공개 데이터셋인 bigquery-public-data.google_analytics_sample을 FastAPI를 통해 외부에서 호출 가능한 API 형태로 변환함으로써, BigQuery 데이터를 서비스 백엔드 수준에서 다루는 과정을 단계별로 살펴보겠습니다.\n\n## 1. 개발 환경 구축 및 GCP 설정\n\nFastAPI와 BigQuery 클라이언트를 실행하기 위한 로컬 개발 환경과 GCP 프로젝트를 설정합니다. Python 가상환경을 만들고 필요한 패키지를 설치한 뒤, BigQuery API가 활성화된 GCP 프로젝트를 준비합니다.\n\n1.1 python 및 라이브러리 설치 https://www.python.org/downloads/ 설치 후 아래 명령어로 버전을 확인합니다.\n\npython –version\n\n1.2 가상환경 생성 및 활성화\n\npython -m venv .venv\n.venv/scripts/activate\n\n1.3 pip 업그레이드 및 라이브러리 설치\n\npython.exe -m pip install --upgrade pip\npip install fastapi[standard] google-cloud-bigquery google-cloud-bigquery-storage pandas python-dotenv pydantic-settings pyarrow\n\n1.4 GCP 프로젝트 생성 https://console.cloud.google.com/ 새 프로젝트를 생성합니다.\n\n1.5 BigQuery API 활성화 API 및 서비스 → 라이브러리 → BigQuery API → 사용 설정\n\n1.6 서비스 계정 생성 IAM 및 관리자 → 서비스 계정 → 서비스 계정 만들기 권한부여: BigQuery 작업 사용자, 데이터 뷰어\n\n1.7 인증 키 발급 생성된 서비스 계정을 클릭 → 키 → 키 추가 → 새 키 만들기 JSON 형식의 비공개 키 파일을 다운로드 후 프로젝트 루트 디렉터리에 저장합니다.\n\n## 2. 데이터 셋 소개\n\nconsole.cloud.google.com/marketplace/product/obfuscated-ga360-data/obfuscated-ga360-data\n\nRest API에 활용할 샘플 데이터는 bigquery-public-data.google_analytics_sample 입니다. Google Merchandise Store의 실제 웹로그 데이터를 기반으로 만들어진 샘플이며, BigQuery에서 누구나 접근할 수 있습니다.\n\nbigquery-public-data.google_analytics_sample.INFORMATION_SCHEMA.COLUMNS\n\n핵심 테이블은 ga_sessions_* 입니다. 날짜 단위로 쪼개져 있으며, 각 세션의 정보가 담겨 있습니다. 스키마는 다소 복잡하지만, 크게 보면 다음과 같은 그룹으로 구성됩니다.\n\n트랜잭션 정보 : 구매 여부, 구매 건수, 매출 금액\n\n디바이스 정보 : 브라우저, 운영체제, 모바일 여부 등\n\n지역 정보 : 국가, 도시, 대륙\n\n행동 정보 : 페이지뷰, 세션 길이 등\n\n## 3. 프로젝트 파일 구조\n\n아래는 FastAPI 애플리케이션의 계층형 구조 예시입니다. 각 디렉터리는 역할별로 모듈화되어 있으며, 의존 방향은 routers → services → repositories → core 순으로 구성됩니다. 현재 예제에서는 app.state.bq를 통해 BigQuery 클라이언트를 전역으로 주입하지만, 규모가 커질 경우 의존성 관리와 리소스 생명주기 제어를 위해 Provider(LOC 컨테이너) 역할을 하는 별도의 모듈이 필요합니다.\n\npytest/\n├── app/\n│ ├── main.py # FastAPI 앱 진입점\n│ │\n│ ├── core/ # BigQuery 클라이언트\n│ │ ├── __init__.py\n│ │ └── bigquery.py\n│ │\n│ ├── models/ # 데이터 모델\n│ │ ├── __init__.py\n│ │ └── ga_sessions.py\n│ │\n│ ├── repositories/ # 데이터 접근 계층\n│ │ ├── __init__.py\n│ │ └── ga_sessions.py\n│ │\n│ ├── services/ # 비즈니스 로직 계층\n│ │ ├── __init__.py\n│ │ └── ga_sessions.py\n│ │\n│ └── routers/ # API 엔드포인트 계층\n│ ├── __init__.py\n│ └── ga_sessions.py\n│\n└── config/ # 설정 파일 (BigQuery 인증 키 파일 위치)\n\n## 4. .env 설정\n\n환경 변수 파일 .env를 통해 BigQuery 자격 증명 경로, 프로젝트 ID, 리전 등의 설정값을 관리합니다. 이 값들은 코드 내에서 os.getenv()로 불러와 클라이언트 초기화에 사용됩니다.\n\nGOOGLE_APPLICATION_CREDENTIALS=\nBIGQUERY_PROJECT_ID=\nBIGQUERY_LOCATION=US\n\n## 5. BigQuery 클라이언트 설정\n\n앱 시작 시 FastAPI의 lifespan 컨텍스트 내에서 BigQuery 클라이언트를 초기화하고, 앱 종료 시 세션을 닫습니다. 이 클라이언트는 Depends를 통해 각 요청 핸들러에서 주입됩니다.\n\n#app/core/bigquery.py\nfrom dotenv import load_dotenv\nfrom contextlib import asynccontextmanager\nfrom google.cloud import bigquery\nimport os\nfrom fastapi import Request\n\nload_dotenv()\n\n@asynccontextmanager\nasync def lifespan(app):\nclient = bigquery.Client.from_service_account_json(\nos.getenv(\"GOOGLE_APPLICATION_CREDENTIALS\"),\nproject=os.getenv(\"BIGQUERY_PROJECT_ID\"),\nlocation=os.getenv(\"BIGQUERY_LOCATION\"),\n)\napp.state.bq = client\nyield\nclient.close()\n\ndef get_bq_client(request: Request):\nreturn request.app.state.bq\n\n## 6. 모델 정의\n\nPydantic 모델을 사용해 BigQuery 쿼리 결과를 구조화합니다. 필요한 주요 필드만 정의하여 응답 스키마를 간결하게 유지하고, 추가 필드는 무시합니다.\n\n# app/models/ga_sessions.py\nfrom typing import Optional, List, Dict, Any\nfrom pydantic import BaseModel, Field, ConfigDict\n\nclass TotalsSlim(BaseModel):\nvisits: Optional[int] = None\nhits: Optional[int] = None\npageviews: Optional[int] = None\ntimeOnSite: Optional[int] = None\n\nclass TrafficSourceSlim(BaseModel):\nsource: Optional[str] = None\nmedium: Optional[str] = None\nreferralPath: Optional[str] = None\n\nclass DeviceSlim(BaseModel):\nbrowser: Optional[str] = None\noperatingSystem: Optional[str] = None\nisMobile: Optional[bool] = None\n\nclass GeoSlim(BaseModel):\ncontinent: Optional[str] = None\ncountry: Optional[str] = None\n\nclass CustomDimension(BaseModel):\nindex: Optional[int] = None\nvalue: Optional[str] = None\n\nclass GaSessionSlim(BaseModel):\nvisitorId: Optional[str] = Field(None, description=\"INT64 → string\")\nvisitId: Optional[str] = Field(None, description=\"INT64 → string\")\nfullVisitorId: Optional[str] = None\n\nvisitNumber: Optional[int] = None\nvisitStartTime: Optional[int] = None\ndate: Optional[str] = None\nchannelGrouping: Optional[str] = None\n\ntotals: Optional[TotalsSlim] = None\ntrafficSource: Optional[TrafficSourceSlim] = None\ndevice: Optional[DeviceSlim] = None\ngeoNetwork: Optional[GeoSlim] = None\ncustomDimensions: Optional[List[CustomDimension]] = None\n\n# 선언하지 않은 필드는 무시\nmodel_config = ConfigDict(extra=\"ignore\")\n\n## 7. Repository 구현: 공개데이터 조회\n\nRepository 계층은 실제 BigQuery SQL을 실행해 데이터를 가져옵니다. 날짜 포맷 검증을 통해 SQL 인젝션이나 잘못된 요청을 방지하고, 파라미터 바인딩으로 limit을 제어합니다.\n\n# app/repositories/ga_sessions.py\nimport re\nfrom typing import Iterable, Dict, Any, List\nfrom google.cloud import bigquery\n\n_YMD_RE = re.compile(r\"^\\d{8}$\")\n\ndef _validate_yyyymmdd(yyyymmdd: str) -> None:\nif not _YMD_RE.fullmatch(yyyymmdd):\nraise ValueError(\"yyyymmdd must be 8 digits like 20160801\")\n\ndef query_sessions(bq: bigquery.Client, yyyymmdd: str, limit: int) -> Iterable[Dict[str, Any]]:\n_validate_yyyymmdd(yyyymmdd)\ntable = f\"`bigquery-public-data.google_analytics_sample.ga_sessions_{yyyymmdd}`\"\n\nsql = f\"\"\"\nSELECT\nCAST(visitorId AS STRING) AS visitorId,\nCAST(visitId AS STRING) AS visitId,\nfullVisitorId,\nvisitNumber,\nvisitStartTime,\ndate,\nchannelGrouping,\ntotals.visits AS totals_visits,\ntotals.hits AS totals_hits,\ntotals.pageviews AS totals_pageviews,\ntotals.timeOnSite AS totals_timeOnSite,\ntrafficSource.source AS trafficSource_source,\ntrafficSource.medium AS trafficSource_medium,\ntrafficSource.referralPath AS trafficSource_referralPath,\ndevice.browser AS device_browser,\ndevice.operatingSystem AS device_operatingSystem,\ndevice.isMobile AS device_isMobile,\ngeoNetwork.continent AS geoNetwork_continent,\ngeoNetwork.country AS geoNetwork_country,\ncustomDimensions AS customDimensions\nFROM {table}\nLIMIT @limit\n\"\"\"\n\njob_config = bigquery.QueryJobConfig(\nquery_parameters=[bigquery.ScalarQueryParameter(\"limit\", \"INT64\", limit)]\n)\nrows = bq.query(sql, job_config=job_config).result()\nfor r in rows:\nyield {k: r.get(k) for k in r.keys()}\n\n## 8. Service\n\nService 계층은 Repository에서 가져온 flat한 쿼리 결과를 Pydantic 모델 형태로 변환합니다. 이 과정을 통해 API 응답이 일관된 구조(GaSessionSlim)를 유지합니다.\n\n# app/services/ga_sessions.py\nfrom typing import Dict, Any, List\nfrom google.cloud.bigquery import Client\nfrom app.models.ga_sessions import GaSessionSlim\nfrom app.repositories.ga_sessions import query_sessions\n\ndef _nest(record: Dict[str, Any]) -> Dict[str, Any]:\nreturn {\n\"visitorId\": record.get(\"visitorId\"),\n\"visitId\": record.get(\"visitId\"),\n\"fullVisitorId\": record.get(\"fullVisitorId\"),\n\"visitNumber\": record.get(\"visitNumber\"),\n\"visitStartTime\": record.get(\"visitStartTime\"),\n\"date\": record.get(\"date\"),\n\"channelGrouping\": record.get(\"channelGrouping\"),\n\"totals\": {\n\"visits\": record.get(\"totals_visits\"),\n\"hits\": record.get(\"totals_hits\"),\n\"pageviews\": record.get(\"totals_pageviews\"),\n\"timeOnSite\": record.get(\"totals_timeOnSite\"),\n},\n\"trafficSource\": {\n\"source\": record.get(\"trafficSource_source\"),\n\"medium\": record.get(\"trafficSource_medium\"),\n\"referralPath\": record.get(\"trafficSource_referralPath\"),\n},\n\"device\": {\n\"browser\": record.get(\"device_browser\"),\n\"operatingSystem\": record.get(\"device_operatingSystem\"),\n\"isMobile\": record.get(\"device_isMobile\"),\n},\n\"geoNetwork\": {\n\"continent\": record.get(\"geoNetwork_continent\"),\n\"country\": record.get(\"geoNetwork_country\"),\n},\n\"customDimensions\": record.get(\"customDimensions\"),\n}\n\ndef to_ga_sessions(records: List[Dict[str, Any]]) -> List[GaSessionSlim]:\nreturn [GaSessionSlim(**_nest(r)) for r in records]\n\ndef get_ga_sessions(bq: Client, yyyymmdd: str, limit: int) -> List[GaSessionSlim]:\nrecords = list(query_sessions(bq, yyyymmdd, limit))\nreturn to_ga_sessions(records)\n\n## 9. Router 및 요청 예시\n\n라우터는 /ga_sessions/{yyyymmdd} 형태의 엔드포인트를 제공하며, limit 쿼리 파라미터로 결과 개수를 제한합니다.\n\n# app/routers/ga_sessions.py\nfrom typing import List\nfrom fastapi import APIRouter, Depends, HTTPException, Query\nfrom google.cloud.bigquery import Client\nfrom app.core.bigquery import get_bq_client\nfrom app.models.ga_sessions import GaSessionSlim\nfrom app.services.ga_sessions import get_ga_sessions\n\nrouter = APIRouter(prefix=\"/ga_sessions\", tags=[\"ga_sessions\"])\n\n@router.get(\"/{yyyymmdd}\", response_model=List[GaSessionSlim])\ndef list_ga_sessions(\nyyyymmdd: str,\nlimit: int = Query(100, ge=1, le=1000),\nbq: Client = Depends(get_bq_client),\n):\ntry:\nreturn get_ga_sessions(bq, yyyymmdd, limit)\nexcept ValueError as ve:\nraise HTTPException(status_code=400, detail=str(ve))\nexcept Exception as e:\nraise HTTPException(status_code=500, detail=\"Query failed\")\n\n## 10. 실행 및 테스트\n\n아래 명령어로 로컬 서버를 실행한 후 브라우저에서 /ga_sessions/20160801 엔드포인트를 호출해 응답을 확인합니다.\n\nuvicorn app.main:app --reload --host 127.0.0.1 --port 8000\n\n/ga_sessions/20160801 호출 결과\n\n이번 포스트에서는 다음 과정을 통해 BigQuery 공개 데이터셋을 REST API 형태로 제공하는 전체 흐름을 구현했습니다. 공개 데이터셋뿐 아니라 다른 BigQuery 프로젝트에도 동일한 구조를 적용할 수 있으며, 인증·캐싱·비동기 쿼리 처리 등을 추가해 확장할 수도 있습니다.\n\nRef. FastAPI 구글 애널리틱스 샘플\n\n최신 마케팅/고객 데이터 활용 사례를 받아보실 수 있습니다.\n비즈스프링 뉴스레터 구독하기 →\n\n## Related Posts via Categories\n\n구글애즈가 정확검색·구문검색 키워드까지 AI 모드에 노출하기 시작했다 — 자동화 상품 없이 AI 답변 화면에 들어가는 첫 실험\nAI가 React 앱을 직접 디버깅 – Agent 기반 디버깅\nChatGPT 광고가 픽셀·전환 API·맞춤 오디언스를 갖췄다 — 답변 화면이 측정 가능한 광고 매체가 되는 순간\nAI 시대, 데이터도 설명이 필요합니다\nMCP 클라이언트는 Authorization Server를 어떻게 찾는가: 공식 MCP TypeScript SDK와 VS Code 구현 비교\nRAG 시대의 GEO: AI가 콘텐츠를 읽는 방식과 스키마의 진짜 역할\nAI 에이전트로 웹 데이터 수집 가능 여부 자동 검증하기\nGA4 Intraday 실시간 테이블, 대시보드 원천 데이터로 바로 사용할 수 있을까?\n워크플로우 오케스트레이션 플랫폼 Temporal 알아보기\niOS 14.5부터 GA4까지, 환경 변화가 가져온 업무 폭증을 해결하는 실무 전략은…?\n\n다음에 대해 검색하기...\n\n최신 글 둘러보기\n\n기존 SEO와 무엇이 같고 무엇이 다른가\n\n콘텐츠가 답인 걸 모르는 사람은 없습니다 — 문제는 지속입니다\n\n광고비를 늘리지 않고 매출을 늘린 회사들은 무엇을 했나\n\n코호트로 비교하면 광고 예산 판단이 달라집니다\n\nGEO를 위한 글쓰기를 위한 작은 노하우\n\n(해외동향) 광고주가 광고플랫폼 리포트에서 자체 측정으로 옮겨가고 있다.\n\n구글애즈가 정확검색·구문검색 키워드까지 AI 모드에 노출하기 시작했다 — 자동화 상품 없이 AI 답변 화면에 들어가는 첫 실험\n\nAI가 React 앱을 직접 디버깅 – Agent 기반 디버깅\n\nChatGPT 광고가 픽셀·전환 API·맞춤 오디언스를 갖췄다 — 답변 화면이 측정 가능한 광고 매체가 되는 순간\n\n마케팅 자동화, 무엇부터 자동화해야 할까?\n\n출처: https://blog.bizspring.co.kr/%ed%85%8c%ed%81%ac/fastapi%ec%99%80-bigquery-%ea%b3%b5%ea%b0%9c%eb%8d%b0%ec%9d%b4%ed%84%b0%eb%a1%9c-rest-api-%eb%a7%8c%eb%93%a4%ea%b8%b0/",
  "embedding": null,
  "embedded_at": null,
  "embedding_hash": null,
  "map_x": null,
  "map_y": null,
  "backlinks": [],
  "html_url": "https://bizspring.ai/kb/ref/web/blog-bizspring-co-kr/fastapi와-bigquery-공개데이터로-rest-api-만들기-bizspring-blog",
  "markdown_url": "https://bizspring.ai/kb/ref/web/blog-bizspring-co-kr/fastapi와-bigquery-공개데이터로-rest-api-만들기-bizspring-blog.md"
}