Google Drive¶
Reference for GoogleKit Drive (googlekit.gdrive) — a typed client for
Google Drive API v3.
Official Google docs: Drive API guides · REST reference · Files · Permissions · Changes · Enable API
Site: https://ssujitx.github.io/GoogleKit/
Overview¶
Install and enable the API:
In Google Cloud Console, enable the Google Drive API for your project and create OAuth client credentials (desktop) and/or a service-account key as needed.
GoogleKit Drive exposes four managers on a single client, plus optional flat shortcuts (same operations, simpler scripts):
| Manager | Attribute | Role |
|---|---|---|
FilesManager |
drive.files |
List, search, upload, download, export, metadata, trash, delete |
FoldersManager |
drive.folders |
Create folders/paths, list children, recursive directory sync |
PermissionsManager |
drive.permissions |
Share with users/groups/domain/anyone, roles, shareable links |
ChangesManager |
drive.changes |
Start page token and incremental changes feed |
Optional shortcuts vs managers¶
Both appear in IDE autocomplete after drive. (typed as DriveAPI). Prefer managers for advanced options; use shortcuts for everyday scripts.
| Shortcut | Delegates to |
|---|---|
list_files(...) |
files.list(...) |
list_folders(...) |
files.list with folder MIME filter |
search_files(name) / search_folders(name) |
files.search(...) |
create_folder(name, parent_id=...) |
folders.create(...) |
upload_file(path, folder_id=...) |
files.upload_path(..., parents=[...]) |
download_file(file_id, destination=...) |
files.download_path(...) |
upload_folder(local_path, ...) |
folders.upload_directory(...) |
delete_file / delete_folder |
files.trash or files.delete |
share / unshare / list_permissions |
permissions.* |
get_share_link(..., public=True) |
permissions.create_shareable_link |
from googlekit.auth.scopes import ScopeProfile
from googlekit.gdrive import DriveClient
drive = DriveClient.from_oauth("client_secret.json", profile=ScopeProfile.FULL)
# Manager
page = drive.files.list(folder_id="root")
# Shortcut (equivalent)
page = drive.list_files(folder_id="root")
folder = drive.create_folder("Reports") # or drive.folders.create(...)
result = drive.upload_file("a.pdf", folder_id=folder.id)
drive.share(result.file.id, email="you@example.com")
Typical entry points:
Or construct the Drive client directly:
from googlekit.gdrive import DriveClient
drive = DriveClient.from_oauth("client_secret.json", token_path="token.json")
Auth & scopes¶
See also Authentication (four methods: ADC, service account, OAuth, auto) and Scopes.
1. Application Default Credentials¶
2. Service account¶
from googlekit.gdrive import DriveClient
drive = DriveClient.from_service_account(
"service-account.json",
subject=None, # set for Workspace domain-wide delegation
)
Ordinary service accounts do not see a personal user's Drive. Share files/folders with the service-account email, or use domain-wide delegation with subject.
3. OAuth (desktop)¶
from googlekit import GoogleKit
from googlekit.auth.scopes import ScopeProfile
client = GoogleKit.from_oauth(
"client_secret.json",
token_path="token.json",
services=["gdrive"],
profile=ScopeProfile.READWRITE, # default
)
drive = client.drive
Or via DriveClient:
from googlekit.gdrive import DriveClient
from googlekit.auth.scopes import ScopeProfile
drive = DriveClient.from_oauth(
"client_secret.json",
token_path="token.json",
profile=ScopeProfile.FULL, # full drive scope when needed
)
4. Auto-detect (auto())¶
Separate from from_adc() — discovers ADC or a local credential JSON.
Order: try ADC → look for credential files in the working directory → raise AuthenticationError with guidance.
Drive scope presets¶
ScopeProfile |
OAuth scope | Typical use |
|---|---|---|
metadata |
drive.metadata.readonly |
List/metadata only |
readonly |
drive.readonly |
Read files without write |
readwrite (default) |
drive.file |
Files created/opened by the app |
full |
drive |
Broad access (empty trash, full My Drive, etc.) |
Prefer drive.file (READWRITE) when possible. Operations like empty_trash() require the full drive scope (ScopeProfile.FULL).
Custom scopes:
from googlekit.auth.scopes import Scope, ScopeSet
from googlekit.gdrive import DriveClient
drive = DriveClient.from_oauth(
"client_secret.json",
scopes=ScopeSet.of(Scope.DRIVE_READONLY),
)
DriveClient¶
Constructors¶
| Method | Signature |
|---|---|
from_oauth |
(client_secrets: str \| Path, token_path: str \| Path \| None = None, scopes: ScopeSet \| list[str] \| None = None, *, profile: ScopeProfile = ScopeProfile.READWRITE, config: ClientConfig \| None = None) -> DriveClient |
from_service_account |
(credentials_file: str \| Path, subject: str \| None = None, scopes: ScopeSet \| list[str] \| None = None, *, profile: ScopeProfile = ScopeProfile.READWRITE, config: ClientConfig \| None = None) -> DriveClient |
from_adc |
(quota_project_id: str \| None = None, scopes: ScopeSet \| list[str] \| None = None, *, profile: ScopeProfile = ScopeProfile.READWRITE, config: ClientConfig \| None = None) -> DriveClient |
from_provider |
(provider: CredentialProvider, *, config: ClientConfig \| None = None) -> DriveClient |
Properties¶
| Property | Type | Description |
|---|---|---|
provider |
CredentialProvider |
Credential provider in use |
config |
ClientConfig |
Runtime config (timeouts, chunk size, Shared Drive flags, retries) |
transport |
Transport |
HTTP/API transport |
files |
FilesManager |
File operations |
folders |
FoldersManager |
Folder / tree operations |
permissions |
PermissionsManager |
Sharing |
changes |
ChangesManager |
Changes feed |
Via the unified client:
from googlekit import GoogleKit
client = GoogleKit.auto(services=["gdrive"])
drive = client.drive # DriveClient
Models¶
Public models from googlekit.gdrive:
| Type | Purpose |
|---|---|
DriveFile |
File/folder metadata (id, name, mime_type, checksums, links, is_folder, is_google_native, …) |
DriveFolder |
Folder metadata |
Permission |
Sharing permission entry |
Change |
One changes-feed entry (removed, file_id, nested file, …) |
UploadResult |
file, bytes_uploaded, overwritten |
DownloadResult |
size, path, mime_type, exported, optional data |
OverwritePolicy |
ERROR / SKIP / OVERWRITE |
PermissionRole |
owner, organizer, fileOrganizer, writer, commenter, reader |
Default metadata field sets used by the managers:
- Files:
id, name, mimeType, parents, size, md5Checksum, …(FILE_FIELDS) - Permissions:
id, type, role, emailAddress, …(PERMISSION_FIELDS) - Changes: includes
nextPageToken,newStartPageToken, and nested file fields (CHANGE_FIELDS)
FilesManager¶
Access: drive.files.
Default page size is 100 (DEFAULT_PAGE_SIZE). List/search return Page[DriveFile]; iterate returns a lazy PageIterator[DriveFile].
list¶
drive.files.list(
*,
query: str | None = None,
folder_id: str | None = None,
page_size: int = 100,
page_token: str | None = None,
order_by: str | None = None,
fields: str = FILE_FIELDS,
corpora: str | None = None,
drive_id: str | None = None,
include_trashed: bool = False,
) -> Page[DriveFile]
One page of files. Trashed items are excluded unless include_trashed=True. When folder_id is set, the query includes '{folder_id}' in parents.
page = drive.files.list(folder_id="root", page_size=50, order_by="name")
for f in page.items:
print(f.name, f.id)
if page.has_more:
next_page = drive.files.list(folder_id="root", page_token=page.next_page_token)
Shared Drive listing: pass drive_id (sets corpora="drive" automatically). corpora="drive" without drive_id raises ValidationError.
iterate¶
drive.files.iterate(
*,
query: str | None = None,
folder_id: str | None = None,
page_size: int = 100,
page_token: str | None = None,
order_by: str | None = None,
fields: str = FILE_FIELDS,
corpora: str | None = None,
drive_id: str | None = None,
include_trashed: bool = False,
) -> PageIterator[DriveFile]
Lazily walks all pages:
search¶
drive.files.search(
query: str,
*,
page_size: int = 100,
page_token: str | None = None,
order_by: str | None = None,
fields: str = FILE_FIELDS,
corpora: str | None = None,
drive_id: str | None = None,
) -> Page[DriveFile]
Drive query syntax (non-empty query required):
get¶
create¶
drive.files.create(
name: str,
*,
mime_type: str = "application/octet-stream",
parents: list[str] | None = None,
fields: str = FILE_FIELDS,
**metadata,
) -> DriveFile
Creates metadata (empty binary file or Google-native doc). Extra kwargs are merged into the API body.
doc = drive.files.create(
"Notes",
mime_type="application/vnd.google-apps.document",
parents=["FOLDER_ID"],
)
copy¶
drive.files.copy(
file_id: str,
*,
name: str | None = None,
parents: list[str] | None = None,
fields: str = FILE_FIELDS,
) -> DriveFile
move¶
drive.files.move(
file_id: str,
*,
add_parents: list[str] | None = None,
remove_parents: list[str] | None = None,
fields: str = FILE_FIELDS,
) -> DriveFile
Requires add_parents and/or remove_parents (raises ValidationError otherwise):
rename¶
update_metadata¶
Patches metadata. Accepts snake_case aliases such as mime_type, modified_time, drive_id, md5_checksum. At least one field is required.
trash / restore¶
drive.files.trash(file_id: str, *, fields: str = FILE_FIELDS) -> DriveFile
drive.files.restore(file_id: str, *, fields: str = FILE_FIELDS) -> DriveFile
empty_trash¶
Permanently deletes all trashed files for the authenticated user. Requires the full drive scope. Pass drive_id to empty trash on a Shared Drive.
delete¶
Permanent delete (cannot be undone):
find_by_name¶
drive.files.find_by_name(
name: str,
*,
parents: list[str] | None = None,
mime_type: str | None = None,
) -> DriveFile | None
First non-trashed exact name match under the given parents:
upload_path¶
drive.files.upload_path(
path: str | Path,
*,
parents: list[str] | None = None,
name: str | None = None,
mime_type: str | None = None,
overwrite: OverwritePolicy | str = OverwritePolicy.ERROR,
chunk_size: int | None = None,
progress: ProgressCallback | None = None,
resumable: bool = True,
) -> UploadResult | None
Uploads a local file. Returns None when overwrite=SKIP and a same-named file already exists. MIME is guessed from the path when omitted.
from googlekit.gdrive import OverwritePolicy
result = drive.files.upload_path(
"report.pdf",
parents=["FOLDER_ID"],
overwrite=OverwritePolicy.OVERWRITE,
)
print(result.file.id, result.overwritten)
upload_bytes¶
drive.files.upload_bytes(
data: bytes,
name: str,
*,
parents: list[str] | None = None,
mime_type: str = "application/octet-stream",
overwrite: OverwritePolicy | str = OverwritePolicy.ERROR,
chunk_size: int | None = None,
progress: ProgressCallback | None = None,
) -> UploadResult | None
result = drive.files.upload_bytes(
b"hello",
name="hello.txt",
mime_type="text/plain",
parents=["FOLDER_ID"],
)
upload_fileobj¶
drive.files.upload_fileobj(
fileobj: BinaryIO,
name: str,
*,
parents: list[str] | None = None,
mime_type: str = "application/octet-stream",
overwrite: OverwritePolicy | str = OverwritePolicy.ERROR,
chunk_size: int | None = None,
progress: ProgressCallback | None = None,
size: int | None = None,
resumable: bool = True,
) -> UploadResult | None
with open("data.bin", "rb") as fh:
drive.files.upload_fileobj(fh, name="data.bin", parents=["FOLDER_ID"], size=1024)
download_path¶
drive.files.download_path(
file_id: str,
destination: str | Path,
*,
chunk_size: int | None = None,
progress: ProgressCallback | None = None,
export_format: str | None = None,
) -> DownloadResult
Streams to disk (does not buffer the whole file). Google-native files require export_format (or use export()).
download_bytes¶
drive.files.download_bytes(
file_id: str,
*,
chunk_size: int | None = None,
progress: ProgressCallback | None = None,
export_format: str | None = None,
) -> DownloadResult
Loads into memory — prefer download_path for large files:
download_fileobj¶
drive.files.download_fileobj(
file_id: str,
fileobj: BinaryIO,
*,
chunk_size: int | None = None,
progress: ProgressCallback | None = None,
export_format: str | None = None,
) -> DownloadResult
export¶
drive.files.export(
file_id: str,
export_format: str,
destination: str | Path | None = None,
*,
chunk_size: int | None = None,
progress: ProgressCallback | None = None,
) -> DownloadResult
Exports a Google-native file. Short names (pdf, docx, xlsx, …) or full MIME types are accepted. With destination=None, returns bytes in DownloadResult.data.
drive.files.export("DOC_ID", "pdf", "notes.pdf")
drive.files.export("SHEET_ID", "xlsx", "data.xlsx")
raw = drive.files.export("DOC_ID", "txt") # bytes in raw.data
FoldersManager¶
Access: drive.folders.
create¶
drive.folders.create(
name: str,
*,
parent_id: str | None = None,
fields: str = FILE_FIELDS,
) -> DriveFolder
create_path¶
drive.folders.create_path(
path: str,
*,
parent_id: str | None = None,
fields: str = FILE_FIELDS,
) -> DriveFolder
Creates nested folders from a slash-separated path, reusing existing names:
list_children¶
drive.folders.list_children(
folder_id: str = "root",
*,
include_folders: bool = True,
include_files: bool = True,
include_trashed: bool = False,
page_size: int = 100,
) -> list[DriveFile]
Eager list of immediate children (use files.iterate for lazy pagination):
upload_directory¶
drive.folders.upload_directory(
local_path: str | Path,
*,
parent_id: str | None = None,
overwrite: OverwritePolicy | str = OverwritePolicy.ERROR,
progress: ProgressCallback | None = None,
create_root: bool = True,
) -> DriveFolder
Recursively uploads a local directory tree. With create_root=True (default), creates a Drive folder named after the local directory. With create_root=False, parent_id is required and files land directly under that folder. Symlink cycles are skipped.
from googlekit.gdrive import OverwritePolicy
dest = drive.folders.upload_directory(
"./project",
parent_id="FOLDER_ID",
overwrite=OverwritePolicy.SKIP,
)
print(dest.id)
download_directory¶
drive.folders.download_directory(
folder_id: str,
destination: str | Path,
*,
overwrite: OverwritePolicy | str = OverwritePolicy.ERROR,
progress: ProgressCallback | None = None,
max_depth: int = 50,
) -> Path
Recursively downloads a Drive folder. Shortcuts are skipped. Google-native Docs/Sheets/Slides are exported as PDF ({name}.pdf) by default. Raises ValidationError if max_depth is exceeded.
PermissionsManager¶
Access: drive.permissions.
Roles (PermissionRole or string): owner, organizer, fileOrganizer, writer, commenter, reader.
list¶
share_user¶
drive.permissions.share_user(
file_id: str,
email: str,
*,
role: PermissionRole | str = PermissionRole.READER,
notify: bool = True,
message: str | None = None,
) -> Permission
from googlekit.gdrive import PermissionRole
drive.permissions.share_user(
"FILE_ID",
"alice@example.com",
role=PermissionRole.WRITER,
message="Please review",
)
share_group¶
drive.permissions.share_group(
file_id: str,
email: str,
*,
role: PermissionRole | str = PermissionRole.READER,
notify: bool = True,
message: str | None = None,
) -> Permission
share_anyone¶
drive.permissions.share_anyone(
file_id: str,
*,
role: PermissionRole | str = PermissionRole.READER,
allow_file_discovery: bool = False,
public: bool = False,
) -> Permission
Creates an anyone permission. Requires public=True — without it, raises ValidationError to prevent accidental public sharing.
share_domain¶
drive.permissions.share_domain(
file_id: str,
domain: str,
*,
role: PermissionRole | str = PermissionRole.READER,
allow_file_discovery: bool = False,
) -> Permission
change_role¶
drive.permissions.change_role(
file_id: str,
permission_id: str,
role: PermissionRole | str,
) -> Permission
remove¶
create_shareable_link¶
drive.permissions.create_shareable_link(
file_id: str,
*,
role: PermissionRole | str = PermissionRole.READER,
public: bool = False,
) -> str
Ensures an anyone permission for the role (if missing) and returns webViewLink. Requires public=True.
ChangesManager¶
Access: drive.changes. Use for incremental sync.
get_start_page_token¶
Save this token before your first sync window. Optional drive_id scopes to a Shared Drive.
list¶
drive.changes.list(
page_token: str,
*,
page_size: int = 100,
drive_id: str | None = None,
include_removed: bool = True,
fields: str = CHANGE_FIELDS,
) -> Page[Change]
One page of changes. When the feed is caught up, the raw response includes newStartPageToken (iteration stops; read it from page.raw).
page = drive.changes.list(token)
for change in page.items:
if change.removed:
print("removed", change.file_id)
elif change.file:
print("updated", change.file.name)
iterate¶
drive.changes.iterate(
page_token: str,
*,
page_size: int = 100,
drive_id: str | None = None,
include_removed: bool = True,
fields: str = CHANGE_FIELDS,
) -> PageIterator[Change]
Lazily consumes the feed until exhausted. Persist newStartPageToken from the last page's raw for the next run (see Incremental changes).
Export formats¶
Short names resolve via EXPORT_MIME_MAP in googlekit.gdrive.export_formats (official Drive export table).
Google Docs (application/vnd.google-apps.document)¶
| Short name | MIME type |
|---|---|
pdf |
application/pdf |
docx |
application/vnd.openxmlformats-officedocument.wordprocessingml.document |
odt |
application/vnd.oasis.opendocument.text |
rtf |
application/rtf |
txt |
text/plain |
html / zip |
application/zip (HTML export is a zip archive) |
epub |
application/epub+zip |
md / markdown |
text/markdown |
Google Sheets (application/vnd.google-apps.spreadsheet)¶
| Short name | MIME type |
|---|---|
pdf |
application/pdf |
xlsx |
application/vnd.openxmlformats-officedocument.spreadsheetml.sheet |
ods |
application/vnd.oasis.opendocument.spreadsheet |
csv |
text/csv |
tsv |
text/tab-separated-values |
html / zip |
application/zip |
Google Slides (application/vnd.google-apps.presentation)¶
| Short name | MIME type |
|---|---|
pdf |
application/pdf |
pptx |
application/vnd.openxmlformats-officedocument.presentationml.presentation |
odp |
application/vnd.oasis.opendocument.presentation |
txt |
text/plain |
png |
image/png |
jpeg / jpg |
image/jpeg |
svg |
image/svg+xml |
Drawings (application/vnd.google-apps.drawing)¶
| Short name | MIME type |
|---|---|
pdf |
application/pdf |
png |
image/png |
jpeg / jpg |
image/jpeg |
svg |
image/svg+xml |
Apps Script¶
| Source MIME | Short name | Export MIME |
|---|---|---|
application/vnd.google-apps.script |
json |
application/vnd.google-apps.script+json |
Invalid combinations raise ValidationError.
Google Vids are not compatible with files.export. Google requires the Drive
long-running files.download operation for MP4 downloads; GoogleKit currently
raises ValidationError instead of issuing an invalid export request.
Shared drives¶
ClientConfig.supports_all_drives defaults to True. When enabled, managers send supportsAllDrives=True, and list/changes list calls also send includeItemsFromAllDrives=True.
from googlekit.core.configuration import ClientConfig
from googlekit.gdrive import DriveClient
drive = DriveClient.from_oauth(
"client_secret.json",
config=ClientConfig(supports_all_drives=True), # default
)
# List files in a Shared Drive
page = drive.files.list(drive_id="SHARED_DRIVE_ID")
# Changes for a Shared Drive
token = drive.changes.get_start_page_token(drive_id="SHARED_DRIVE_ID")
Rules:
- Passing
drive_idtofiles.list/files.iteratesetscorpora="drive"if you omitcorpora corpora="drive"withoutdrive_idraisesValidationErrorempty_trash(drive_id=...)empties trash for that Shared Drive
Disable Shared Drive params when you only need My Drive:
drive = DriveClient.from_oauth(
"client_secret.json",
config=ClientConfig(supports_all_drives=False),
)
Overwrite policies¶
OverwritePolicy controls name collisions on upload and directory sync:
| Value | Behavior |
|---|---|
ERROR (default) |
Raise ValidationError if the destination name already exists |
SKIP |
Leave the existing file; upload helpers return None |
OVERWRITE |
Replace content of the existing Drive file (same id) |
Accepts the enum or string ("error", "skip", "overwrite").
from googlekit.gdrive import OverwritePolicy
drive.files.upload_path("a.pdf", parents=["FOLDER"], overwrite=OverwritePolicy.SKIP)
drive.folders.upload_directory("./data", overwrite="overwrite")
For uploads, existence is checked with find_by_name under parents (or ["root"] when parents are omitted). For download_directory, the policy applies to local paths.
Progress callbacks¶
Type: ProgressCallback = Callable[[int, int | None], None] — (bytes_so_far, total_or_None).
Default chunk size is 256 * 1024 (256 KiB), overridable via chunk_size= or ClientConfig.chunk_size.
def on_progress(done: int, total: int | None) -> None:
if total:
print(f"{done}/{total} ({100 * done / total:.0f}%)")
else:
print(f"{done} bytes")
drive.files.upload_path("large.bin", progress=on_progress)
drive.files.download_path("FILE_ID", "large.bin", progress=on_progress)
Callback exceptions are logged and do not abort the transfer.
Errors¶
Drive operations use the shared hierarchy documented in Errors:
| Situation | Exception |
|---|---|
Bad local input (empty id, missing path, invalid role, share_anyone without public=True) |
ValidationError |
| HTTP 404 | NotFoundError |
| HTTP 409 / 412 | ConflictError |
| HTTP 429 | RateLimitError |
| Quota / 403 quota reasons | QuotaExceededError |
| Missing OAuth scope | InsufficientScopesError |
| Retries exhausted | RetryExhaustedError |
Missing googleapiclient extra |
MissingExtraError |
from googlekit.core.exceptions import ValidationError, NotFoundError
try:
drive.files.get("missing")
except NotFoundError as exc:
print(exc.status_code, exc.reason)
Full recipes¶
Upload a folder¶
from googlekit import GoogleKit
from googlekit.gdrive import OverwritePolicy
client = GoogleKit.auto(services=["gdrive"])
drive = client.drive
dest = drive.folders.upload_directory(
"./project",
parent_id=None, # My Drive root
overwrite=OverwritePolicy.SKIP,
create_root=True,
)
print("Uploaded to folder", dest.id)
Upload into an existing Drive folder without creating a new root folder:
drive.folders.upload_directory(
"./project",
parent_id="EXISTING_FOLDER_ID",
create_root=False,
overwrite=OverwritePolicy.OVERWRITE,
)
Sync a folder tree both ways¶
from googlekit import GoogleKit
from googlekit.gdrive import OverwritePolicy
client = GoogleKit.auto(services=["gdrive"])
drive = client.drive
# Local → Drive
drive.folders.upload_directory(
"./workspace",
parent_id="REMOTE_FOLDER_ID",
create_root=False,
overwrite=OverwritePolicy.OVERWRITE,
)
# Drive → Local (Google Docs/Sheets/Slides become PDF)
drive.folders.download_directory(
"REMOTE_FOLDER_ID",
"./workspace-mirror",
overwrite=OverwritePolicy.OVERWRITE,
)
Create a public link¶
from googlekit import GoogleKit
client = GoogleKit.auto(services=["gdrive"])
drive = client.drive
uploaded = drive.files.upload_path("report.pdf")
assert uploaded is not None
# Explicit public=True is required (safety guard)
link = drive.permissions.create_shareable_link(uploaded.file.id, public=True)
print(link)
# Or create the anyone permission yourself, then read webViewLink
drive.permissions.share_anyone(uploaded.file.id, role="reader", public=True)
meta = drive.files.get(uploaded.file.id, fields="webViewLink")
print(meta.web_view_link)
Empty trash¶
Requires ScopeProfile.FULL (https://www.googleapis.com/auth/drive):
from googlekit.gdrive import DriveClient
from googlekit.auth.scopes import ScopeProfile
drive = DriveClient.from_oauth(
"client_secret.json",
profile=ScopeProfile.FULL,
)
drive.files.empty_trash()
Incremental changes¶
Prefer ScopeProfile.FULL or READONLY so the feed covers more than app-created files. Persist newStartPageToken from the last page's raw dict (use list in a loop when you need that field; iterate does not expose page raw).
from pathlib import Path
from googlekit.gdrive import DriveClient
from googlekit.auth.scopes import ScopeProfile
drive = DriveClient.from_oauth("client_secret.json", profile=ScopeProfile.FULL)
token_path = Path("drive_changes_token.txt")
if token_path.exists():
token = token_path.read_text(encoding="utf-8").strip()
else:
token = drive.changes.get_start_page_token()
token_path.write_text(token, encoding="utf-8")
last_raw: dict | None = None
while True:
page = drive.changes.list(token)
for change in page.items:
if change.removed:
print("deleted", change.file_id)
elif change.file:
print("changed", change.file.id, change.file.name)
last_raw = page.raw
if not page.next_page_token:
break
token = page.next_page_token
new_token = last_raw.get("newStartPageToken") if last_raw else None
if new_token:
token_path.write_text(str(new_token), encoding="utf-8")
Search, move, and export¶
from googlekit import GoogleKit
client = GoogleKit.auto(services=["gdrive"])
drive = client.drive
page = drive.files.search("name contains 'Q1' and trashed = false")
for f in page.items:
if f.is_google_native:
drive.files.export(f.id, "pdf", f"{f.name}.pdf")
else:
archive = drive.folders.create_path("Archive/2026")
drive.files.move(f.id, add_parents=[archive.id], remove_parents=f.parents)